数据库集成

FastAPI 不内置 ORM，生态标准是 SQLAlchemy 2.0 + Alembic，驱动按同步/异步选 psycopg3 或 asyncpg。面试主线：会话生命周期怎么管理、同步异步怎么选、迁移怎么做。

SQLAlchemy 2.0 模型

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(unique=True)

2.0 风格：Mapped[...] 类型注解驱动，声明式基类
字段类型、约束、索引都在列定义里

会话生命周期：yield 依赖

会话随请求创建销毁, 连接从池中借还

from sqlalchemy.orm import sessionmaker

engine = create_engine(DATABASE_URL, pool_size=10, max_overflow=20)
SessionLocal = sessionmaker(bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
        db.commit()          # 无异常提交
    except Exception:
        db.rollback()        # 异常回滚
        raise
    finally:
        db.close()           # 归还连接

一个请求一个 Session：yield 依赖创建，请求结束提交/回滚/关闭
连接池：pool_size + max_overflow 控制；池满时请求等待（超时配 pool_timeout）
Session 不是线程安全的，每请求新建是正确姿势

同步 vs 异步驱动

组合	适用
SQLAlchemy 同步 + psycopg3	默认选择：简单、生态全、配 def 路由
SQLAlchemy 异步 + asyncpg	高并发 IO：async def 路由，全程 async
asyncpg 裸驱动	极致性能，手写 SQL

# 异步版
engine = create_async_engine(DATABASE_URL, pool_size=10)   # postgresql+asyncpg://
async def get_db():
    async with AsyncSessionLocal() as db:
        yield db

选型结论：默认同步 + psycopg3（FastAPI 自动丢线程池，够用且简单）；QPS 很高或大量 IO 等待时上 asyncpg。同步 SQLAlchemy 配 def 路由，异步配 async def，别混。

Alembic 迁移

alembic init alembic              # 初始化
alembic revision -m "create users table"   # 生成迁移脚本
alembic upgrade head              # 应用迁移

迁移脚本记录 schema 变更历史（建表、加列、改索引）
alembic upgrade 应用到目标版本，downgrade 回滚
生成脚本要人工检查（自动生成不完美，尤其索引和约束）
生产流程：本地生成 → 审查 → 部署时 upgrade

面试追问

Session 为什么每请求新建？ Session 非线程安全，且要控制事务边界。yield 依赖创建销毁，连接归还池
连接池参数？ pool_size（常驻连接）、max_overflow（峰值追加）、pool_timeout（等待超时）
同步还是异步驱动？ 默认同步 psycopg3 + def 路由（简单够用）；高并发 IO 用 asyncpg + async def
Alembic 干什么？ schema 版本管理：迁移脚本记录变更，upgrade/downgrade 应用回滚
事务边界在哪？ 依赖的 yield 后统一 commit，异常 rollback。路由里不散落 commit