我有一个带有异步 sqlalchemy 的异步 FastApi 应用程序，源代码：
from sqlalchemy import (
Column,
String,
)
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.decl_api import DeclarativeMeta
from app.config import settings
engine = create_async_engine(settings.DATABASE_URL, pool_per_ping=True)
Base: DeclarativeMeta = declarative_base()
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
class Titles(Base):
__tablename__ = "titles"
id = Column(String(100), primary_key=True)
title = Column(String(100), unique=True)
async def get_session() -> AsyncSession:
async with async_session() as session:
yield session
import .database
from fastapi_utils.cbv import cbv
from …我有一个带有异步 sqlalchemy 的异步 FastApi 应用程序，源代码（不会提供 schemas.py，因为没有必要）：
from sqlalchemy import (
Column,
String,
)
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.decl_api import DeclarativeMeta
from app.config import settings
engine = create_async_engine(settings.DATABASE_URL)
Base: DeclarativeMeta = declarative_base()
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
class Titles(Base):
__tablename__ = "titles"
id = Column(String(100), primary_key=True)
title = Column(String(100), unique=True)
async def get_session() -> AsyncSession:
async with async_session() as session:
yield session
import .database
from fastapi_utils.cbv import cbv
from …