第 10 章 数据库会话与事务

学习目标

学完本章，你应该能：

1. 理解什么是数据库会话（Session）和事务（Transaction）
2. 掌握 fba 中 CurrentSession 和 CurrentSessionTransaction 的用法
3. 知道"查询"和"增删改"分别该用哪个

一、什么是会话和事务

先搞懂两个核心概念：

• 会话（Session）：你的代码和数据库之间的"对话通道"。通过它执行查询、增删改操作。
• 事务（Transaction）：一组要么全部成功、要么全部回滚的操作。比如转账：扣 A 的钱 + 加 B 的钱，必须一起成功或一起失败，不能只成功一半。

回滚（rollback）：事务失败时，撤销所有已做的操作，回到事务开始前的状态。

事务的原子性示意

二、fba 的两种会话依赖

fba 用 FastAPI 的依赖注入，提供了两种现成的会话：

1. CurrentSession —— 用于查询

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

CurrentSession = Annotated[AsyncSession, Depends(get_db)]

用法（用于查询操作）：

@router.get('')
async def get_user_list(db: CurrentSession) -> ResponseModel:
    users = await user_dao.get_list(db)  # 查询
    return response_base.success(data=users)

2. CurrentSessionTransaction —— 用于增删改

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

CurrentSessionTransaction = Annotated[AsyncSession, Depends(get_db_transaction)]

用法（用于增删改操作）：

@router.post('')
async def create_user(obj: CreateUserParam, db: CurrentSessionTransaction) -> ResponseModel:
    user = await user_dao.create(db, obj)  # 自动在事务中执行
    return response_base.success(data=user)

区别：CurrentSessionTransaction 会调用 .begin() 自动开启事务，增删改完成后自动提交（commit）。

关键区别对照

对比项	CurrentSession	CurrentSessionTransaction
用途	查询	增删改
是否自动开事务	否	是（.begin()）
是否自动提交	否	是

如何选择会话

三、begin() 方式（灵活控制事务）

除了依赖注入，fba 还支持在任意位置（service、工具函数）手动控制事务：

async def create(*, obj: CreateUserParam) -> None:
    async with async_db_session.begin() as db:
        await user_dao.create(db, obj)
        # 如果需要，可以在这里执行多个操作，作为一个事务
        await other_dao.update(db, ...)

注意：这种方式需要开发者自行控制事务边界，避免在同一业务流程中重复开启不必要的事务。

四、关于"查询也开事务"的说明

官方文档专门解释了一个容易困惑的点：为什么查询语句看起来也开了事务？

简单说：这是 SQLAlchemy（以及大多数 Python 数据库连接器）的默认行为——任何遵循 PEP 249 规范的连接器都会默认开启事务。这是正常的，不是 bug，不需要担心。

PEP 249：Python 数据库 API 规范。简单理解就是"Python 访问数据库的标准约定"。

五、一个完整的增删改查示例

综合第 05-10 章的知识，看一个完整的用户模块：

# api 层
@router.get('/{user_id}', summary='查询用户')
async def get_user(user_id: int, db: CurrentSession) -> ResponseModel:
    user = await user_service.get_user_detail(db, user_id)
    return response_base.success(data=user)

@router.post('', summary='创建用户')
async def create_user(obj: CreateUserParam, db: CurrentSessionTransaction) -> ResponseModel:
    user = await user_service.create_user(db, obj)
    return response_base.success(data=user)

# service 层
class UserService:
    async def get_user_detail(self, db, user_id: int):
        return await user_dao.get(db, user_id)

    async def create_user(self, db, obj: CreateUserParam):
        return await user_dao.create(db, obj)

# crud 层
class UserCRUD(BaseCRUD[User]):
    pass

user_dao = UserCRUD(User)

核心规律：

• 查询接口 → db: CurrentSession
• 增删改接口 → db: CurrentSessionTransaction

六、常见疑问

Q1：忘记用 CurrentSessionTransaction 会怎样？ 增删改操作可能不会自动提交，导致数据没真正写入数据库（或行为不一致）。所以写操作务必用带事务的会话。

Q2：什么时候用 begin() 手动方式？ 当多个数据库操作需要作为一个整体（要么都成功要么都回滚）时，比如"创建订单 + 扣库存"必须同时成功，这时手动用 begin() 把它们包在一个事务里更安全。

本章小结

关键要点回顾：

• 会话是代码与数据库的对话通道，事务保证一组操作原子性
• CurrentSession 用于查询，CurrentSessionTransaction 用于增删改
• 也可用 async_db_session.begin() 手动控制事务
• 查询"也开事务"是正常现象，不用纠结

到这里，你已经掌握了 fba 的核心开发流程。 从第 11 章开始进入进阶能力（配置、权限、缓存、异步任务等）。

---

对应官方文档：事务（backend/reference/transaction.html）