FastAPI 学习教程走到第5部分终于要碰真实项目里绕不开的那堵墙了——数据持久化。前四部分我们把路由、依赖注入、Pydantic 校验玩得还算熟练但所有数据都是内存里的临时对象一重启服务就什么都没了。从这一部分开始我会带你把 FastAPI 应用接到数据库上用 SQLAlchemy 2.0 的异步 ORM 管理表结构并顺手解决迁移、会话控制这些让人头大的问题。这篇文章最适合那些已经能写出接口但不知道怎么把数据真正存下来的开发者。1. 为什么第5部分要讲数据库一个真实项目的演进1.1 前四部分结束后留下的问题我们在前四部分一直在做一个模拟的待办事项项目接口从/ping一直写到了带路径参数的GET /items/{id}响应模型能自动校验、依赖注入能复用 token 校验逻辑。但说实话那个项目的所有数据都存在一个进程内的 Python 列表里只要服务一重启新增的任务、登录状态全部清空。更麻烦的是一旦想加入用户体系不同用户之间的数据就无法隔离因为所有任务都挤在一个全局列表里。这是所有 API 服务迟早要面对的问题没有持久化层业务逻辑就只能是玩具。当时我在给某内部工具升级时也遇到了同样的瓶颈。接口已经写好了测试也通过但部署到服务器上跑不了几天内存就被占满了。后来才意识到不能再用列表即数据库的方案必须引入真正的数据库。FastAPI 自身只是个 Web 框架它不管数据存储只给你一个灵活的空间让你接入任何你习惯的 ORM。因此在选择数据库访问层时我最看重的是成熟度、异步支持、迁移工具三个点。1.2 ORM选型为什么是SQLAlchemy 2.0市面上的 Python ORM 不少Tortoise ORM 用起来很 Pythonic直接对着模型类操作但文档里很多异步细节得自己试用原生 SQL 又太原始每次写字段映射要命。我最终选择了 SQLAlchemy 2.0原因很直接第一它在 FastAPI 官方文档里长期作为示例方案社区出现问题时最容易搜到答案第二它同时提供同步和异步两套 API当你还在调试阶段可以用同步模式跑得更快需要上线时切到异步模式改动成本小第三它配套的 Alembic 迁移工具在表结构演进上几乎是标准答案后面要改字段、加约束都不会手忙脚乱。有人可能会问第5部分是不是太早接触 ORM 了我的看法是越早越好。数据库模型是整个业务逻辑的地基你越早把模型定义清楚后面的路由、校验、测试就能越扎实。而且 FastAPI 本身的依赖注入体系非常适合搭配 SQLAlchemy 的 Session你会看到它们俩配合得天衣无缝这也是本部分最核心的内容。2. 工程结构重组与最小依赖集2.1 目录结构与配置正式接入数据库前我建议先把项目结构梳理一下否则后面模型、路由、迁移脚本会全部塞进main.py改一处牵连一片。常见的做法是按下层、模型、接口拆开我目前使用的结构是app/ main.py # 创建 FastAPI 实例、注册路由 core/ database.py # 引擎、会话、Base 全部在这里 models/ __init__.py # 集中导出所有模型 user.py item.py schemas/ user.py # Pydantic 请求/响应模型 item.py api/ routes/ users.py items.py这个结构在小型项目中可能显得稍微重但当你从第5部分开始加入数据库后你会发现每个模块都有了清晰的家。core/database.py是整个数据层的入口models只关心表结构schemas负责接口校验api/routes只写业务路由。这样分工之后后面加迁移脚本、加测试都会很舒服。2.2 依赖安装与版本陷阱安装依赖时我踩过一个不大不小的坑直接把sqlalchemy装上然后继续用同步方式操作结果在异步路由里调用session.query()时卡了很久。正确的做法是根据数据库类型安装对应驱动程序。下面是我常用的最小依赖集pip install fastapi uvicorn[standard] sqlalchemy alembic pydantic pip install asyncpg # PostgreSQL 异步驱动 pip install aiosqlite # SQLite 异步驱动如果你的数据库是 PostgreSQLasyncpg是必须的如果是本地调试想用 SQLiteaiosqlite需要一起装上。两个驱动名字很像但功能完全不同我甚至有一次把asyncpg装到了 SQLite 项目里import 直接报错所以装之前先确认自己的数据库是什么。SQLAlchemy 2.0 对 Python 版本要求是 3.7 以上但异步特性最好用 3.10 及以上我自己在 3.12 上跑得很稳。另外特别提醒一句不要在同一环境里混合安装sqlalchemy和sqlalchemy-async这类第三方兼容包。SQLAlchemy 2.0 原生支持异步无需额外包装装了反而可能造成重复符号冲突。如果你看到网上老教程教你用async_session async_sessionmaker(...)那才是 2.0 的现代写法。3. 模型定义与异步引擎从零搭起数据库骨架3.1 声明式基类与模型映射进入正题。我们先在core/database.py里创建引擎和会话工厂这段代码一定要理解透因为后续所有数据库操作都依赖它。以 PostgreSQL 为例# app/core/database.py from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker from sqlalchemy.orm import DeclarativeBase DATABASE_URL postgresqlasyncpg://user:passwordlocalhost:5432/mydb engine create_async_engine(DATABASE_URL, echoTrue, pool_size10, max_overflow20) async_session async_sessionmaker(engine, expire_on_commitFalse) class Base(DeclarativeBase): pass这里有几个关键点。create_async_engine创建的是异步引擎它内部维护着一个连接池pool_size10表示最少连接数max_overflow20表示最多在基础连接之外额外创建 20 个连接。echoTrue会在控制台打印 SQL 语句调试阶段很有用但上线前记得改成 False。expire_on_commitFalse非常关键否则每次commit()后之前拿到的对象属性都会过期再次访问时会触发隐式的数据库查询在异步模式下会直接报MissingGreenlet错误这个我们后面专门说。模型定义方面我以用户表和项目表为例# app/models/user.py from datetime import datetime from sqlalchemy import String, DateTime, func from sqlalchemy.orm import Mapped, mapped_column from app.core.database import Base class User(Base): __tablename__ users id: Mapped[int] mapped_column(primary_keyTrue) email: Mapped[str] mapped_column(String(255), uniqueTrue, indexTrue) hashed_password: Mapped[str] mapped_column(String(255)) is_active: Mapped[bool] mapped_column(defaultTrue) created_at: Mapped[datetime] mapped_column(DateTime(timezoneTrue), server_defaultfunc.now())SQLAlchemy 2.0 的Mapped[mapped_column()]写法比旧版的Column更简洁而且类型标注和数据库类型天然结合IDE 能正确推断出类型。注意indexTrue可以为高频查询字段建索引uniqueTrue保证邮箱不重复这些都是数据库级约束比在应用层判断靠谱得多。3.2 异步引擎与会话工厂的生命周期引擎应该是模块级单例一个应用启动时只创建一次。不要在每次请求里都创建引擎否则连接池会反复创建销毁导致资源浪费。会话工厂async_session也是单例通过它获取会话。每次业务操作都应该通过async with async_session() as session:来开启一个会话。生命周期上引擎跟随应用进程存活会话则应该是每次请求或每次后台任务一个实例用完即关。FastAPI 的依赖注入正好能帮我们管理这个生命周期下一节会细说。但在此之前你要清楚一点async_sessionmaker创建的是会话工厂它本身不是会话。这有点像一家餐厅有多个服务员会话而餐厅本身是工厂每次叫人来服务就递给你一个服务员服务完了服务员回岗位。4. 在路由层安全使用数据库会话4.1 用依赖注入替代手动会话管理如果你在每个路由里手动async def创建会话、try/finally 关闭代码很快就重复得让人崩溃。FastAPI 依赖注入提供了非常优雅的方案# app/core/database.py from typing import AsyncGenerator async def get_session() - AsyncGenerator[AsyncSession, None]: async with async_session() as session: yield session然后在路由中使用from fastapi import Depends, APIRouter from sqlalchemy.ext.asyncio import AsyncSession router APIRouter() router.get(/users/{user_id}) async def get_user(user_id: int, session: AsyncSession Depends(get_session)): ...为什么这样好因为 FastAPI 会在请求进入时调用get_session()会话通过async with创建路由中通过await执行查询。请求处理完后FastAPI 会执行get_session后面没有被yield的代码——这里是退出async with自动关闭会话。这个模式把创建、使用、关闭的脏活全部收编到框架手里你只需要关注业务逻辑。4.2 CRUD实操注册用户与查询文章现在我们动手写几个真实接口。注册用户的逻辑from sqlalchemy import select from passlib.hash import bcrypt # 别再直接存明文密码 router.post(/register, status_code201) async def register_user(user_data: UserCreate, session: AsyncSession Depends(get_session)): existing await session.scalar(select(User).where(User.email user_data.email)) if existing: raise HTTPException(status_code400, detail邮箱已注册) user User( emailuser_data.email, hashed_passwordbcrypt.hash(user_data.password), is_activeTrue, ) session.add(user) await session.commit() await session.refresh(user) return user语法上await session.scalar()是 SQLAlchemy 2.0 的常用手段它会执行查询并返回第一行结果或 None。session.add()加入新对象commit()刷新到数据库。注意我这里用了refresh(user)把数据库默认填充的created_at时间取回来否则刚创建的对象里created_at是空的。查询列表带分页也很简单router.get(/users) async def list_users(offset: int 0, limit: int 20, session: AsyncSession Depends(get_session)): result await session.execute(select(User).order_by(User.id).offset(offset).limit(limit)) users result.scalars().all() return users这里有个经验不要直接返回result.scalars()的迭代器而是用.all()转成列表否则 FastAPI 在序列化时可能会再次访问未加载的数据导致异步上下文问题。更新和删除分别是await session.execute(update(User).where(...))和await session.delete(user)模式相同不再赘述。5. 表结构演进是刚需接入Alembic做迁移5.1 Alembic初始化与异步适配模型定义好之后不可能永远不修改。今天加一个字段明天删一个唯一约束如果每次都把表删了重建线上数据就没了。Alembic 专门解决这个问题它可以生成增量迁移脚本记录表结构的每一次变化。在项目根目录执行alembic init alembic然后改alembic/env.py让它读取我们应用里的模型和数据库 URL。如果你使用异步引擎需要额外配置alembic.ini里的sqlalchemy.url或者直接在env.py里引用DATABASE_URL。最关键的是让env.py能导入我们的模型否则 autogenerate 会根据空数据库生成迁移导致表结构丢失。在env.py顶部加import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent)) from app.core.database import Base from app.models import User, Item # noqa target_metadata Base.metadata这样 Alembic 就知道你有哪些模型了。然后修改迁移运行方式官方 2.0 版本提供了async_main示例但大部分情况下直接在env.py里使用同步 URL 也能跑迁移因为迁移本身不需要高并发。我个人的做法是保留一个同步引擎专门给 Alembic比如sqlalchemy.create_engine(DATABASE_URL.replace(asyncpg, ))如果你用 PostgreSQL去掉驱动前缀即可。5.2 迁移脚本的生成与执行第一次迁移可以用自动生成alembic revision --autogenerate -m create users and items tables alembic upgrade head--autogenerate会对比当前数据库状态和target_metadata的差异生成 Python 迁移脚本。注意它不会处理表重命名、字段重命名这类需要数据迁移的操作这些复杂场景下我们还是得手动改生成的脚本文档。迁移脚本生成后建议立刻打开看一眼确认op.create_table里的字段类型和我们的模型一致。我之前遇到一次String(255)变成了String(length255)的迁移其实没问题但如果你发现字段类型不匹配趁升级之前改脚本最安全。执行完alembic upgrade head数据库里就有alembic_version表了里面记录了当前数据库的版本号。以后每次模型变化执行一次 autogenerate 再 upgrade表结构就能平滑演进。这里有个实操技巧让表结构变更跟随代码分支一起提交每个人本地环境都跑一遍迁移就不会出现我的库比你的库多两个字段这种问题了。6. 我在第5部分反复踩的四个坑6.1 同步驱动出现在异步上下文第一个崩溃场景是我在异步路由里用了session.execute()却连接的是用了同步驱动的数据库 URL。症状就是运行到查询语句时直接抛出MissingGreenlet错误或者提示 greenlet_spawn has not been called。这个问题本质是SQLAlchemy 的异步 API 底层依赖 greenlet 协程来切换数据库线程如果你给create_async_engine传了一个同步 URL比如postgresql://而不是postgresqlasyncpg://异步方法尝试从同步连接池拿连接时greenlet 就无法正确启动。解决办法简单粗暴检查DATABASE_URL中是否写了正确的驱动前缀。PostgreSQL 用postgresqlasyncpg://SQLite 用sqliteaiosqlite://。如果你不小心混着写在调试时看到MissingGreenlet第一时间检查 URL。6.2 SQLite内存库在测试中的假故障项目早期我图方便测试时用sqliteaiosqlite:///:memory:结果所有测试并行运行时每个请求拿到的其实是不同的内存数据库。因为 SQLite 内存数据库在每个连接关闭时会清空而 FastAPI 依赖注入创建的会话是每次请求一个几个测试并发时会话可能连到不同的内存库数据自然是各查各的。后来我改成了sqliteaiosqlite:///./test.db或者用不以:memory:开头的临时文件路径测试瞬间就稳定了。如果你一定想测内存模式可以在所有测试进程共用一个全局会话但那样就失去并行测试的意义了。本地开发用 SQLite 没问题上线换 PostgreSQL 之前记得在 CI 里跑一遍 PostgreSQL 环境下的测试否则有些 SQL 方言差异会在生产里冒出来。6.3 会话耗尽与连接池参数在默认参数下SQLAlchemy 异步引擎的连接池大小是 5最大溢出是 10。如果你的服务并发稍高而每个请求内还会执行多次数据库查询可能瞬间把连接占满后续请求全部排队接口响应时间飙到好几秒。我当时在压测一个小接口发现 QPS 上不去数据库日志里全是waiting for connection。排查后我给create_async_engine加上了pool_size20, max_overflow20并把echoFalse关闭日志瞬间就好了。这里要注意连接池大小并不是越大越好它受制于数据库后端的最大连接数你把应用层设成 50数据库可能只允许 20 个连接结果照样排队。更合理的做法是给数据库连接池设置合理的上限同时给慢查询加索引。6.4 N1查询让接口变慢当我们用 ORM 表示关联关系时最经典的坑是 N1 查询。比如查询所有项目时需要拿到每个项目的所属用户信息如果按懒加载逐个访问project.owner就会产生 1 条项目查询 N 条用户查询。在异步场景下这个现象尤其隐蔽因为性能问题不会立刻暴露直到数据量上来。解决办法是在查询时显式声明关联加载from sqlalchemy.orm import selectinload result await session.execute( select(Project).options(selectinload(Project.owner)) ) projects result.scalars().all()selectinload会额外发一条SELECT ... WHERE id IN (...)的语句把关联对象一次性载入从而避免逐条访问。这个优化在项目列表中特别明显效果可以从几百毫秒降到几十毫秒。你还可以用joinedload去执行 JOIN 查询不过一对多关联时使用selectinload往往更不容易产生数据重复的副作用。第5部分内容到这里其实已经覆盖了数据库接入最核心的骨架。如果只能留三句话我会说存储层提前规划、会话必须用依赖注管理、迁移脚本跟随代码走。最后再分享一个小技巧在开发阶段保持echoTrue让数据库把每条 SQL 打到控制台你会对 ORM 的行为和性能产生非常重要的直觉等上线前再关掉它。