1. 连接资源

FastAPI 接入数据库时，需要先区分 Session 的生命周期和事务的生命周期：

1
应用级：Engine、连接池、Session 工厂
2
请求任务级：AsyncSession
3
业务动作级：数据库事务

把这三层放进一次真实请求中，执行顺序会更直观：

1
HTTP 请求
2
  -> FastAPI 为本次请求创建 AsyncSession
3
  -> 路由校验参数，并调用 Service
4
  -> Service 开启事务
5
  -> Repository 执行 SQL，flush 获得数据库生成值
6
  -> 事务提交或回滚
7
  -> 路由把 ORM 对象转换成 Pydantic 响应对象
8
  -> FastAPI 关闭 AsyncSession

Engine 不是“一条始终打开的数据库连接”，而是连接池及连接获取逻辑的入口。AsyncSession 通常在第一次执行 SQL 时才从连接池取得连接，事务结束后再把连接归还给连接池。因此，应用可以共享 Engine，请求之间却不能共享 Session。

Engine 应在应用进程中复用。如果每次请求都创建 Engine，就会同时创建很多连接池，失去复用连接的意义。使用多个 worker 时，每个进程都有自己的 Engine 和连接池，因此还要结合 worker 数量规划 PostgreSQL 的总连接上限。

1
from sqlalchemy.ext.asyncio import create_async_engine
2

3
from app.core.settings import settings
4

5

6
engine = create_async_engine(
7
    settings.database_url,
8
    pool_pre_ping=True,
9
)

pool_pre_ping=True 会在从连接池取出连接时检查它是否仍然可用，能够减少数据库重启或网络中断后拿到失效连接的情况，但也会增加一次连接检查。是否设置连接池大小、溢出数量和等待超时，应根据部署进程数与数据库容量决定。

Session 工厂同样可以全局复用，但它创建出来的 Session 不能全局共享：

01
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
02

03
from app.db.engine import engine
04

05

06
session_factory = async_sessionmaker(
07
    engine,
08
    class_=AsyncSession,
09
    expire_on_commit=False,
10
)

应用关闭时应释放 Engine 持有的连接。使用 FastAPI lifespan：

01
from collections.abc import AsyncIterator
02
from contextlib import asynccontextmanager
03

04
from fastapi import FastAPI
05

06
from app.db.engine import engine
07

08

09
@asynccontextmanager
10
async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
11
    try:
12
        yield
13
    finally:
14
        await engine.dispose()
15

16

17
app = FastAPI(lifespan=lifespan)

try/finally 保证应用退出或 lifespan 内部出现异常时都会释放 Engine 持有的连接。这里没有调用 Base.metadata.create_all()，数据库结构由部署阶段的 Alembic 迁移准备，应用启动只负责使用现有结构。

Session 依赖

FastAPI 的 yield 依赖适合管理一次请求的 Session 生命周期：

01
from collections.abc import AsyncIterator
02
from typing import Annotated
03

04
from fastapi import Depends
05
from sqlalchemy.ext.asyncio import AsyncSession
06

07
from app.db.session import session_factory
08

09

10
async def get_db_session() -> AsyncIterator[AsyncSession]:
11
    async with session_factory() as session:
12
        try:
13
            yield session
14
        except Exception:
15
            await session.rollback()
16
            raise
17

18

19
DbSession = Annotated[
20
    AsyncSession,
21
    Depends(get_db_session, scope="function"),
22
]

请求进入路径函数前会创建 Session，路由和服务共用它。scope 只对包含 yield 的依赖有意义。这里把依赖范围设为 scope="function"，因此路径函数执行完成后、响应发送前就会运行 yield 后的清理代码，Session 随即关闭，连接也会归还给连接池。

这种设置有一个重要前提：路由必须在返回前把 ORM 对象转换成已经包含响应数据的 Pydantic 对象，不能在 Session 关闭后再触发关系加载。流式接口应优先在开始发送响应前读取完所需数据；如果生成器确实要在发送期间继续查询数据库，则需要让 Session 覆盖完整的流式响应生命周期，并控制持续时间，否则慢客户端会长期占用 Session 和连接。

依赖不会自动提交。事务边界由掌握完整业务动作的服务控制，这样一个业务动作可以调用多个仓库方法，最后一起提交或回滚。异常路径中的显式 rollback() 是一道兜底；Session 关闭时也会结束尚未完成的事务，但显式回滚能让异常处理的意图更清楚。只读查询产生的隐式事务会在 Session 关闭时结束。

同一个请求中的顺序调用可以共享 Session，但不能把同一个 AsyncSession 同时传入多个 asyncio 并发任务。每个并发任务都需要自己的 Session 和事务。

DbSession 把类型和依赖声明组合起来，路由参数只需要写一个清楚的类型名。它与前面 FastAPI 依赖篇介绍的 Annotated 是同一种用法。