事务

约 449 字大约 2 分钟

默认情况下，如果将数据库引擎参数 echo 设置为 True，你将会看到事务总是被开启，即便那是一个查询语句。但这并不是因为我们错误的使用了 SQLAlchemy，你可以查看 #6921、#12782 了解详情

简要总结

任何遵循 PEP 249 进行设计的 Python 数据库连接器或 ORM，通常都会默认开启事务

在 SQLAlchemy 中，你可以选择不使用它自身的事务模式，但这需要将数据库本身的事务隔离级别设置为 AUTOCOMMIT ，详情请查看: 了解 DBAPI 级别的 Autocommit 隔离级别

CurrentSession

这是一种接近 FastAPI 官方示例的使用方式。它提供普通数据库会话，通常用于查询操作

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    """获取数据库会话"""
    async with async_db_session() as session:
        yield session
        
# Session Annotated
CurrentSession = Annotated[AsyncSession, Depends(get_db)]

这种方法通常直接应用于接口函数。会话生命周期由依赖注入管理，适合请求级数据库访问

@router.get('')
async def get_pagination_apis(db: CurrentSession) -> ResponseModel:
    ...

CurrentSessionTransaction

与 CurrentSession 不同，此方法会通过 async_db_session.begin() 自动开启事务，适合增删改操作。

async def get_db_transaction() -> AsyncGenerator[AsyncSession, None]:
    """获取带有事务的数据库会话"""
    async with async_db_session.begin() as session:
        yield session

# Session Annotated
CurrentSessionTransaction = Annotated[AsyncSession, Depends(get_db_transaction)]

使用方法与 CurrentSession 相同

@router.post('')
async def create_api(db: CurrentSessionTransaction) -> ResponseModel:
    ...

`begin()`

这种方式由 SQLAlchemy 官方实现，可在 service、工具函数等任意位置使用。与请求依赖方式相比，它需要开发者自行控制调用边界，避免在同一业务流程中重复开启不必要的事务

async def create(*, obj: CreateIns) -> None:
    async with async_db_session.begin() as db:
        await xxx_dao.create(db, obj)