FastAPI-数据库

这篇讲什么

• FastAPI 不绑 ORM，官方姿势是 SQLAlchemy 2.x + 一个 yield 依赖发 session。这篇以 MySQL + mysqlclient 为主路径，并用本机 WSL 的 PostgreSQL 18.6 做异步对照。
• 三条实测结论：异步能力是 sqlalchemy[asyncio] 这个 extra，不装就没有 greenlet，import 直接失败；expire_on_commit=True 在异步下会让 commit 后读属性抛 MissingGreenlet；session 依赖必须 try/finally，否则异常路径不收尾。
• 本地小查询下异步并不比同步快（20 次 select 1：同步 0.041s、异步串行 0.054s、异步并发 0.052s），异步的收益在慢数据库和高并发等待。

装什么

• 同步主路径三个包：

  pip install "sqlalchemy>=2" mysqlclient

  Would install SQLAlchemy-2.1.8 mysqlclient-2.3.0

  • mysqlclient（导入名 MySQLdb）是官方推荐的 MySQL 驱动，C 实现、比纯 Python 快，本机实测：

    create_engine("mysql+mysqldb://root:password@127.0.0.1:3306/fastapi_probe?charset=utf8mb4")
    select version() -> 9.7.4

  • 不想装 C 依赖就用 pymysql，SQLAlchemy 里换方言前缀即可，实测两者都能连同一台库。

• 异步要多一步，这一步最容易漏：

  pip install "sqlalchemy[asyncio]" aiomysql asyncpg

  Would install greenlet-3.5.6 aiomysql-0.3.2 asyncpg-0.32.0 PyMySQL-1.2.3

  • sqlalchemy[asyncio] 带的是 greenlet。只装 sqlalchemy 时，from sqlalchemy.ext.asyncio import create_async_engine 会直接失败：

    ImportError: The SQLAlchemy asyncio module requires that the Python 'greenlet' library
    is installed. In order to ensure this dependency is available, use the
    'sqlalchemy[asyncio]' install target: 'pip install sqlalchemy[asyncio]'

    报错写得很客气，但这是 import 期失败，服务根本起不来，别以为是版本不兼容。

  • aiomysql 会顺手把 PyMySQL 拉进来（它复用了 PyMySQL 的协议实现），所以看到两个包一起装是正常现象。

• MySQL 异步驱动本机实测两个都能跑完整接口：aiomysql（mysql+aiomysql://）和 asyncmy（mysql+asyncmy://），PostgreSQL 用 asyncpg（postgresql+asyncpg://）。

• 连接串对照（口令一律走环境变量，不要把明文写进代码）：

  mysql+mysqldb://USER:password@127.0.0.1:3306/fastapi_probe?charset=utf8mb4
  mysql+pymysql://USER:password@127.0.0.1:3306/fastapi_probe?charset=utf8mb4
  mysql+aiomysql://USER:password@127.0.0.1:3306/fastapi_probe?charset=utf8mb4
  mysql+asyncmy://USER:password@127.0.0.1:3306/fastapi_probe?charset=utf8mb4
  postgresql+psycopg://USER:password@127.0.0.1:5600/fastapi_probe
  postgresql+asyncpg://USER:password@127.0.0.1:5600/fastapi_probe

  • MySQL 一定要带 charset=utf8mb4，否则中文进库变问号，装法与坑见 MySQL入门。
  • PostgreSQL 连不上 5432 是因为本机把它挪到了 5600（Windows 保留端口区间），细节见 PostgreSQL入门。

模型与会话工厂

• SQLAlchemy 2.x 的写法（DeclarativeBase，不再是 declarative_base() 函数）：

  models.py

  from sqlalchemy import Column, DateTime, Integer, String, func
  from sqlalchemy.orm import DeclarativeBase
  
  
  class Base(DeclarativeBase):
      pass
  
  
  class Student(Base):
      __tablename__ = "students"
      id = Column(Integer, primary_key=True)
      name = Column(String(32), unique=True, nullable=False)
      age = Column(Integer, default=0)
      created_at = Column(DateTime, server_default=func.now())

  db.py

  import os
  
  from sqlalchemy import create_engine
  from sqlalchemy.orm import sessionmaker
  
  engine = create_engine(
      os.environ["DATABASE_URL"],
      pool_pre_ping=True,
      pool_size=5,
      max_overflow=10,
      pool_recycle=3600,
  )
  SessionFactory = sessionmaker(bind=engine, expire_on_commit=False, autoflush=False)

  Base.metadata.create_all(engine) 之后
  表结构列名 -> ['id', 'name', 'age', 'created_at']
  写入两行   -> [(1, 'lilei', 21), (2, 'hanmeimei', 20)]
  created_at 由 server_default 填 -> 2026-11-06 21:11:52

  • default= 是 Python 侧填值，server_default= 是建表时写进 DDL 由数据库填；func.now() 走后者，实测拿到的确实是数据库时间。
  • 字段名按 PEP8 用下划线（created_at），别学老笔记写 createdAt。

• 连接池参数不是设完就忘的，能从引擎读回来核对：

  pool 类型: QueuePool | size: 5 | overflow: 10 | recycle: 3600 | pre_ping: True

  • pool_size 是常驻连接数，max_overflow 是峰值时临时多开的数量，两者之和就是数据库端能承受的连接上限。
  • pool_pre_ping=True 会在借出连接前先 ping 一次，专治 MySQL 的 wait_timeout 断连（8 小时空闲被服务端踢掉）。
  • pool_recycle=3600 是连接活多久就重建，配合 pre_ping 用；MySQL 侧的会话时区、字符集在建连时定，改配置要重连才生效。
  • engine.dispose() 之后新连接 id 会变（实测 91 → 92），说明池真的换了一批连接。

session 依赖

• 标准写法就一段，注意 try/finally：

  deps_db.py

  from typing import Annotated
  
  from fastapi import Depends, FastAPI
  from sqlalchemy.orm import Session
  
  app = FastAPI()
  
  
  def get_db():
      db = SessionFactory()
      try:
          yield db
      finally:
          db.close()

  三次请求（含一次 400、一次 404）后 open/close 计数: 4 4

  • 为什么非要 try/finally：裸 yield 的收尾在异常路径不执行，实测 400 时 close 丢了；包了 try/finally 之后 400 和 422 都照样收尾。详见 FastAPI-依赖注入与异常中间件。
  • 一个请求一个 session，别全局共享；引擎/连接池才是全局的（这条与 Tornado 那篇的结论一致，见 Tornado-模板和数据库以及接口的调用顺序）。

• CRUD 接口实测：

  crud_api.py

  class StudentOut(BaseModel):
      model_config = ConfigDict(from_attributes=True)
      id: int
      name: str
      age: int
  
  
  @app.get("/students", response_model=list[StudentOut])
  def list_students(db: Annotated[Session, Depends(get_db)],
                    skip: int = 0, limit: int = 10):
      return db.query(Student).order_by(Student.id).offset(skip).limit(limit).all()
  
  
  @app.get("/students/{sid}")
  def one_student(sid: int, db: Annotated[Session, Depends(get_db)]):
      row = db.get(Student, sid)
      if not row:
          raise HTTPException(404, "没有这个学生")
      return {"id": row.id, "name": row.name}

  GET /students            -> 200 [{'id': 1, 'name': 'lilei', 'age': 21},
                                    {'id': 2, 'name': 'hanmeimei', 'age': 20}]
  GET /students/999        -> 404 {'detail': '没有这个学生'}
  GET /students?skip=1&limit=1 -> [{'id': 2, 'name': 'hanmeimei', 'age': 20}]

  • 直接把 ORM 对象列表交给 response_model=list[StudentOut] 就行，FastAPI 会逐个转。
  • 分页用 offset/limit 够用；深分页要改成「游标分页」（where id > last_id limit n），MySQL 的 offset 100000 会先扫十万行，索引原理见 MySQL入门。

• 唯一约束冲突实测：

  db.add(Student(name="lilei")); db.commit()
  -> IntegrityError (MySQLdb.IntegrityError) (1062, "Duplicate entry 'lilei' for key 'students.name'")
  db.rollback() 之后 session 仍可用 -> 2

  • 不接这个异常，接口就是 500（实测 FastAPI 里未处理的 IntegrityError 直接返回 500）。要给用户友好提示就 except IntegrityError: raise HTTPException(409, "已存在")。

  • 出错后必须先 rollback()，否则 session 处于「脏」状态，实测不 rollback 直接再查就报：

    s.query(Student).count()
    -> PendingRollbackError: This Session's transaction has been rolled back
       due to a previous exception during flush
    s.rollback(); s.query(Student).count() -> 2      # 回滚后 session 照常可用

ORM 对象转 Pydantic

• 关键就是 from_attributes，不开就只能吃字典：

  orm_to_pydantic.py

  class StudentOut(BaseModel):
      model_config = ConfigDict(from_attributes=True)
      id: int
      name: str
      age: int

  StudentOut.model_validate(row, from_attributes=True) -> {'id': 1, 'name': 'lilei', 'age': 21}
  没开 from_attributes -> ValidationError: 1 validation error for Loose

  • 在 FastAPI 里不用手动调：response_model=StudentOut 会自动按属性读取（内部就是这条路径）。
  • 自己 model_validate 时要么开 ConfigDict(from_attributes=True)，要么显式传 from_attributes=True，两个都实测可用。

• 关系字段（relationship）在响应模型里要显式声明，否则不会被序列化；反过来，声明了就等于触发一次额外查询（N+1），排查办法见下面异步一节与 Django-让网站访问更快 的同类思路。

异步写法

• 异步引擎 + 异步 session，接口全程 await：

  async_db.py

  import os
  from contextlib import asynccontextmanager
  from typing import Annotated
  
  from fastapi import Depends, FastAPI, Request
  from sqlalchemy import Column, Integer, String, select
  from sqlalchemy.ext.asyncio import (AsyncSession, async_sessionmaker,
                                      create_async_engine)
  from sqlalchemy.orm import DeclarativeBase
  
  
  class Base(DeclarativeBase):
      pass
  
  
  class Student(Base):
      __tablename__ = "astudents"
      id = Column(Integer, primary_key=True)
      name = Column(String(32), unique=True)
  
  
  @asynccontextmanager
  async def lifespan(app: FastAPI):
      app.state.engine = create_async_engine(os.environ["DATABASE_URL"], pool_size=5)
      app.state.sf = async_sessionmaker(app.state.engine, expire_on_commit=False)
      async with app.state.engine.begin() as conn:
          await conn.run_sync(Base.metadata.create_all)
      yield
      await app.state.engine.dispose()
  
  
  app = FastAPI(lifespan=lifespan)
  
  
  async def get_session(request: Request):
      async with request.app.state.sf() as session:
          yield session
  
  
  @app.get("/a")
  async def list_a(session: Annotated[AsyncSession, Depends(get_session)]):
      rows = (await session.execute(select(Student).order_by(Student.id))).scalars().all()
      return {"names": [r.name for r in rows]}

  POST /a?name=probe-abc -> 200 {'added': 'probe-abc'}
  GET  /a                -> [{'id': 1, 'name': 'async-x'}, {'id': 7, 'name': 'asyncmy-probe'}]

  三个驱动（asyncpg / aiomysql / asyncmy）用同一套代码跑通，实测都返回 200。

• expire_on_commit=False 不是可选项，实测两种配置的差别：

  expire_on_commit=True   commit 后读 row.name -> MissingGreenlet: greenlet_spawn has not
                                                  been called; can't call await_() here
  expire_on_commit=False  commit 后读 row.name -> async-x

  • 原因：默认 commit 后对象属性「过期」，再访问就要回库重取；而异步下这种隐式 IO 不可能同步发生，于是 SQLAlchemy 直接抛 MissingGreenlet。
  • 所以异步 session 一律 async_sessionmaker(..., expire_on_commit=False)，要新数据就重新查。

• 异步下 session.execute() 必须 await，且 SQL 要包 text()：

  await session.execute("select 1")
  -> ArgumentError: Textual SQL expression 'select 1' should be explicitly declared as text('select 1')

  这是 SQLAlchemy 2.0 的硬规矩（1.x 的裸字符串在 2.x 直接报错），跟 FastAPI 无关，但异步代码里更容易写错。

• 引擎别跨事件循环复用。实测在 asyncio.run() 里建好引擎、再交给 TestClient 自己的循环用，会撞上：

  RuntimeError: Task ... got Future ... attached to a different loop

  • 正确做法就是上面那段：引擎在 lifespan 里建、在 lifespan 里 dispose()，它天然活在这个服务唯一的循环里。
  • 测试时也别在模块导入阶段建引擎，用 with TestClient(app) as c: 让 lifespan 先跑起来。

• 异步不等于更快，本机 20 次 select 1（库就在旁边）：

  同步串行 20 次: 0.041s
  异步串行 20 次: 0.054s
  异步并发 20 次: 0.052s

  • 本地亚毫秒查询下异步只有额外开销；把并发拉起来也压不下去，因为瓶颈是网络往返而不是等待队列。
  • 真正划算的场景是「数据库远、单次几十毫秒、并发高」：这时同步要堆线程池（上限 40，见 FastAPI-基本入门），异步一个循环就挂住几千个等待。

迁移用 Alembic

• 官方推荐的迁移工具就是 Alembic（和 SQLAlchemy 同一个作者），实测 1.20.1 可用：

  pip install alembic
  alembic init alembic

  Creating directory ... alembic ...  done
  子命令: check current downgrade history init revision upgrade

• 要改两处配置：alembic.ini 里的 sqlalchemy.url（建议改成从环境变量读），和 alembic/env.py 里的 target_metadata = Base.metadata，否则 autogenerate 看不到你的模型，会生成空迁移。

• 标准三步：

  alembic revision --autogenerate -m "add students"
  alembic upgrade head

  • --autogenerate 是「对比模型与数据库后生成脚本」，生成的脚本要人过一遍：删列、改类型、加索引这类它可能漏判或判错。
  • alembic downgrade -1 回退一步，alembic current 看当前版本，alembic history 看链条。

• 别用 Base.metadata.create_all() 上生产：它只建不改，表结构变了不会同步。开发期图快可以用，一旦有数据就必须走 Alembic。Django 那边有内置迁移，对照写法见 Django-模型。

参考

• FastAPI 官方文档 - SQL Dependencies
• SQLAlchemy 2.0 官方文档 - ORM Quick Start
• SQLAlchemy 2.0 - Asyncio
• SQLAlchemy 2.0 - Pooling
• Alembic 官方文档
• aiomysql / asyncmy / asyncpg