FastAPI数据库连接池配置有效性及代码合理性咨询

问题背景 #

现有FastAPI数据库连接配置及依赖定义如下：

应用启动/数据库连接代码

application.add_event_handler(
    "startup",
    create_start_app_handler(application, settings),
)

def create_start_app_handler(
    app: FastAPI,
    settings: AppSettings,
) -> Callable: 
    async def start_app() -> None:
        await connect_to_db(app, settings)
    return start_app

async def connect_to_db(app: FastAPI, settings: AppSettings) -> None:
    db_url = settings.DATABASE_URL
    engine = create_engine(db_url, pool_size=settings.POOL_SIZE, max_overflow=settings.MAX_OVERFLOW)

    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    db = SessionLocal()

    def close_db():
        db.close()
        engine.dispose()

    app.state.db = db
    app.state.close_db = close_db

依赖定义代码

def _get_db(request: Request) -> Generator:
    yield request.app.state.db

def get_repository(
    repo_type: Type[BaseRepository],
) -> Callable[[Session], BaseRepository]:
    def _get_repo(
        sess: Session = Depends(_get_db),
    ) -> BaseRepository:
        return repo_type(sess)

    return _get_repo

一、是否能利用数据库连接池？ #

不能真正利用连接池的多连接能力。

虽然代码中创建engine时配置了pool_size和max_overflow参数，但启动时只创建了一个全局Session实例并挂载到app.state.db，所有请求都复用这同一个Session。SQLAlchemy的Session会绑定连接池中的一个连接，这个连接会被一直占用直到应用关闭，连接池中的其他连接完全无法被使用，相当于连接池配置形同虚设。

二、实现中的不规范问题及规避点 #

1. 全局单Session的线程安全风险 #

SQLAlchemy的Session本身不是线程/并发安全的，FastAPI是异步框架，会并发处理多个请求，所有请求共用同一个Session会导致：

• 事务交叉执行，数据错乱
• 并发操作时出现锁冲突、会话状态异常

规避方案：每个请求独立创建Session，请求结束后关闭，保证会话隔离。

2. Session生命周期管理错误 #

当前实现把Session的生命周期和应用绑定，从启动到关闭一直存在，违背了SQLAlchemy的最佳实践：Session应该是短生命周期的，对应单个请求或单个业务操作，用完即关闭，将连接归还到连接池。

规避方案：在_get_db依赖中动态创建Session，请求结束时自动关闭，示例：

def _get_db(request: Request) -> Generator[Session, None, None]:
    # 每个请求创建新的Session实例
    db = request.app.state.SessionLocal()
    try:
        yield db
    finally:
        # 请求结束后关闭Session，归还连接到池
        db.close()

3. 应用关闭逻辑不完善 #

当前close_db只关闭了全局Session和销毁engine，但如果是请求级Session，需要保证所有活跃Session都被正确关闭。另外，启动时不应提前创建Session实例，只需要初始化engine和SessionLocal工厂类即可。

修正后的启动/关闭逻辑：

async def connect_to_db(app: FastAPI, settings: AppSettings) -> None:
    db_url = settings.DATABASE_URL
    engine = create_engine(db_url, pool_size=settings.POOL_SIZE, max_overflow=settings.MAX_OVERFLOW)
    # 只保存engine和Session工厂类，不创建实例
    app.state.engine = engine
    app.state.SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def create_shutdown_app_handler(app: FastAPI) -> Callable:
    async def shutdown_app() -> None:
        # 应用关闭时销毁engine，回收连接池资源
        app.state.engine.dispose()
    return shutdown_app

# 注册关闭事件
application.add_event_handler("shutdown", create_shutdown_app_handler(application))

4. 异步与同步代码混用的阻塞风险 #

connect_to_db是异步函数，但内部使用了同步版SQLAlchemy的create_engine和Session，在FastAPI的异步事件循环中执行同步数据库操作会阻塞整个服务，导致性能下降。

规避方案：如果用异步FastAPI，建议改用SQLAlchemy的异步版本（create_async_engine、AsyncSession），配合异步数据库驱动（如asyncpg）。

内容的提问来源于stack exchange，提问作者SamAko