事务
约 441 字大约 1 分钟
默认情况下,如果将数据库引擎参数 echo 设置为 True,你将会看到事务总是被开启,即便那是一个查询语句。但这并不是因为我们错误的使用了 SQLAlchemy,你可以查看 #6921、#12782 了解详情
简要总结
任何遵循 PEP-429 进行设计的 Python 数据库连接器或 ORM,都将默认开启事务
在 SQLAlchemy 中,你可以选择不使用它自身的事务模式,但这需要将数据库本身的事务隔离级别设置为 AUTOCOMMIT ,详情请查看: 了解 DBAPI 级别的 Autocommit 隔离级别
CurrentSession
这是一种类似于官方文档的使用方法,但这种方法并没有真正开启事务,它通常仅用于查询操作
async def get_db() -> AsyncGenerator[AsyncSession, None]:
"""获取数据库会话"""
async with async_db_session() as session:
yield session
# Session Annotated
CurrentSession = Annotated[AsyncSession, Depends(get_db)]这种方法通常直接应用于接口函数,在 session 应用方面,它被认为是线程安全的
@router.get('')
async def get_pagination_apis(db: CurrentSession) -> ResponseModel:
...CurrentSessionTransaction
与 CurrentSession 不同,此方法将直接自动开启事务,你可以将它用于增删改操作
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 get_pagination_apis(db: CurrentSession) -> ResponseModel:
...begin()
这种方式由 SQLAlchemy 官方实现,在线程安全方面,由于在同一个函数中,可能存在多次调用,所以没有 CurrentSession 和 CurrentSessionTransaction 更加严谨,但此方式可以在任意地方使用
async def create(*, obj: CreateIns) -> None:
async with async_db_session.begin() as db:
await xxx_dao.create(db, obj)

