三种获取 AsyncSession 方式方式 1异步上下文管理器推荐直接使用AsyncSessionFactory()配合async with, session退出自动关闭内部嵌套async with.begin()手动开启事物正常结束自动commit抛出异常自动rollbackfrom pydantic import BaseModel from models import AsyncSessionFactory, User from fastapi import HTTPException # 返回响应Schema class UserRespSchema(BaseModel): id: int email: str username: str class Config: orm_mode True # 请求Schema class UserCreateReqSchema(BaseModel): email: str username: str password: str app.post(/user/add, response_modelUserRespSchema) async def add_user(req: UserCreateReqSchema): # 创建session退出自动关闭session async with AsyncSessionFactory() as session: try: # begin()手动开启事务正常退出自动commit异常自动rollback async with session.begin(): user User( usernamereq.username, emailreq.email, passwordreq.password ) session.add(user) # 不需要手动写 await session.commit() except Exception as e: # 事务上下文已经完成回滚直接抛异常即可 raise HTTPException(status_code400, detail用户名或邮箱已经存在) return user方式 2FastAPI 依赖注入获取 Session把session创建封装 为yield依赖视图函数通过Depends注入sessionfinlly保证请求结束关闭session。适合项目同意管理session业务路由直接拿session使用from fastapi import Depends from models import AsyncSessionFactory, AsyncSession async def get_session(): session AsyncSessionFactory() try: yield session finally: await session.close() app.post(/user/add, response_modelUserRespSchema) async def add_user( req: UserCreateReqSchema, session: AsyncSession Depends(get_session) ): try: async with session.begin(): user User(usernamereq.username, emailreq.email, passwordreq.password) session.add(user) except Exception: raise HTTPException(status_code400, detail用户名或邮箱已经存在) return user方式 3HTTP 中间件绑定 session 到 request.state在全局中间件创建 session 挂载request.state.session所有路由可以从 request 对象拿 session。缺点全局请求都会创建 session即使接口不需要数据库会浪费连接池资源实际开发很少优先选用。from fastapi import Request app.middleware(http) async def create_session_middleware(request: Request, call_next): session AsyncSessionFactory() # 将session挂载到本次请求state对象 setattr(request.state, session, session) response await call_next(request) # 请求处理完成关闭session await session.close() return response # 路由中使用 app.post(/user/add, response_modelUserRespSchema) async def add_user(req: UserCreateReqSchema, request: Request): session: AsyncSession request.state.session try: async with session.begin(): user User(usernamereq.username, emailreq.email, passwordreq.password) session.add(user) except Exception: raise HTTPException(status_code400, detail用户名或邮箱已经存在) return user事务重点说明async with session.begin()事务上下文代码块正常结束自动 commit不用写await session.commit()代码抛出异常自动 rollback 回滚不用手动写await session.rollback()如果不使用session.begin()需要手动操作session AsyncSessionFactory() user User(...) session.add(user) await session.commit() # 手动提交 #出错时 await session.rollback() await session.close()三种方案对比总结方案优点缺点推荐度async with 上下文管理器简单直观自动关闭 session、事务自动提交回滚无额外配置每个接口内部写创建代码⭐⭐⭐⭐⭐首选Depends 依赖注入路由干净统一 session 逻辑复用性强需要写依赖函数⭐⭐⭐⭐Middleware 中间件全部路由都能拿到 session所有请求强制创建 session浪费连接资源⭐⭐
