FastAPI中SQLAlchemy UOW与Starlette Admin会话冲突报错排查

这个错误的核心原因是异步SQLAlchemy会话不支持并发操作，Starlette Admin在加载关联子模型时，会在同一个请求内触发多个并发数据库查询，而你的UOW如果在整个请求周期内复用同一个会话，就会导致多个任务同时操作同一个会话，触发冲突。另外连接池未归还的问题，通常是会话没有被正确关闭或完成回滚/提交流程。

1. 确保每个数据库操作使用独立会话（关键） #

异步SQLAlchemy的AsyncSession是任务隔离的，不能在多个并发任务中共享。修改你的UOW实现，不要在请求级别复用同一个会话，而是为每个需要数据库操作的任务创建独立实例：

# 错误示例：请求级别单例会话，会导致并发冲突
@app.get("/")
async def get_data(uow: UOW = Depends(get_uow)):
    # 多个并发任务共用同一个uow.session

# 正确做法：每次获取新的UOW实例，用上下文管理器自动管理会话生命周期
async def get_uow():
    async with AsyncSession(engine) as session:
        async with session.begin():
            yield UOW(session=session)

2. 修复Starlette Admin的会话使用逻辑 #

Starlette Admin处理关联模型时会自动触发额外查询，你需要自定义Admin视图，为每个查询操作提供独立会话，避免和业务逻辑会话冲突：

from starlette_admin.contrib.sqla import ModelView

class CustomModelView(ModelView):
    async def get_object(self, pk: Any, request: Request) -> Any:
        # 为加载对象及关联子模型创建独立会话
        async with AsyncSession(engine) as session:
            return await session.get(self.model, pk)
    
    # 同理重写get_list、get_form等需要加载关联数据的方法

3. 完善UOW的会话生命周期管理 #

确保UOW在使用完毕后，无论成功还是失败，都正确执行提交/回滚，不要手动调用session.close()，依赖async with上下文管理器自动回收：

class UOW:
    def __init__(self, session: AsyncSession):
        self.session = session
    
    async def commit(self):
        await self.session.commit()
    
    async def rollback(self):
        await self.session.rollback()

# 业务层使用示例
async def some_service(uow: UOW):
    try:
        # 执行数据库操作
        await uow.commit()
    except Exception:
        await uow.rollback()
        raise

4. 调整SQLAlchemy连接池配置 #

如果连接池未归还问题仍存在，调整连接池参数确保连接被正确回收：

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_pre_ping=True,  # 自动检测并丢弃无效连接
    pool_recycle=300,    # 自动回收闲置超过5分钟的连接
    pool_size=10,        # 根据业务并发量调整
    max_overflow=20
)

5. 排查未捕获异常导致的会话泄漏 #

如果请求过程中出现未被捕获的异常，可能导致会话无法正常关闭。添加全局异常处理器统一处理：

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    # 获取当前请求的UOW实例，确保回滚会话
    uow = request.state.uow
    await uow.rollback()
    return JSONResponse(status_code=500, content={"detail": "Internal Server Error"})

1. 启动应用后，打开Starlette Admin的编辑页面，检查是否还出现会话并发错误
2. 终止应用时，查看控制台是否还有连接池未归还的警告
3. 可以通过SQLAlchemy工具监控连接池状态：

from sqlalchemy import inspect

async def check_pool_status():
    pool = inspect(engine).pool
    print(f"已借出连接数: {pool.checkedout()}")
    print(f"闲置连接数: {pool.idle()}")

内容的提问来源于stack exchange，提问作者Музика Андрій