FastAPI数据操作实战：从ORM选型到会话管理与性能优化

写接口这两年我最大的体会是一个 FastAPI 项目从能跑变成能上线中间隔着的往往就是数据操作这一层。前面几篇讲了路由、参数校验、依赖注入那些都还属于“怎么把请求接进来”的问题到了这一篇我们要解决的是“数据到底怎么存、怎么查、怎么改”的核心问题。这篇是 FastAPI 入门教程系列的第七篇也是我一直觉得最绕不开的一篇。标题叫“数据操作”说白了就是数据库的增删改查但真正落到工程里它涵盖的东西比单纯写 SQL 要多得多ORM 选型、会话管理、模型设计、关系映射、事务边界、连接池、分页性能……随便哪一环没处理好线上都会给你颜色看。这篇文章适合已经把 FastAPI 基础路由、参数、依赖看了一遍、正准备把项目跟数据库接起来的读者也适合已经写了几个接口但总觉得数据层写得别扭、想系统梳理一遍的朋友。1. 数据层设计先想清楚再动手很多新手一上来就急着写接口结果项目写到一半发现模型关系乱成一团、会话到处开、连接池被耗尽。其实数据层是最适合“先设计后编码”的部分因为一旦表结构、关系、会话方式定了后面改动的成本非常高。1.1 为什么我还是选了 SQLAlchemyFastAPI 本身不绑定任何 ORM官方文档里推荐过 SQLAlchemy也确实是最主流的选择。你可能问过现在不是有 Tortoise ORM、PonyORM或者干脆写原生 SQL 吗为什么大多数项目还是选 SQLAlchemy我的答案是SQLAlchemy 的生态太成熟了。Alembic 做数据库迁移、Pydantic 和 SQLAlchemy 模型互转都有现成套路SQLAlchemy 2.0 的Mapped写法让类型标注完整到连 IDE 都能给你做代码提示而且就算你哪天觉得 ORM 不够用了SQLAlchemy 还能退回 Core 层写贴近原生 SQL 的查询不至于被框架卡死。换个角度看你花在 SQLAlchemy 上的学习成本几乎可以用在所有 Python 后端项目里而不是只对 FastAPI 有效。当然如果你是那种“SQL 写得极溜、项目又小、不想学 ORM”的开发者直接用asyncpg或者aiomysql写裸 SQL 也完全可行。只是这种方案的样板代码会多一些表结构一多维护成本就上去了。我个人建议大多数人直接选 SQLAlchemy 2.0 Alembic这是投入产出比最高的组合。1.2 数据层的目录结构怎么摆才不乱我在网上看到很多人问 FastAPI 项目目录结构怎么组织这里给一个经历过几个线上项目考验的布局也是我目前在用的模板app/ ├── main.py # FastAPI 实例、路由注册 ├── core/ │ ├── config.py # 配置读取数据库 URL 等 │ └── database.py # 引擎、会话工厂、Base 定义 ├── models/ # SQLAlchemy 模型 │ ├── user.py │ └── article.py ├── schemas/ # Pydantic 模型请求/响应用 │ ├── user.py │ └── article.py ├── crud/ # 数据库操作层 │ ├── user.py │ └── article.py ├── api/ │ └── v1/ │ ├── user.py # 路由 │ └── article.py很多人会困惑 models 和 schemas 是不是重复了其实分工很明确models是数据库表结构的映射字段类型要和数据库对齐schemas是对外 API 数据格式的定义负责从接口层“翻译”数据。这一层隔离非常重要——你永远不会想因为改了一个 API 返回字段就牵连到数据库表结构。crud这一层是我后来加上的。一开始我觉得 FastAPI 路由里直接写 SQLAlchemy 查询挺方便结果路由越来越长最后把查询逻辑、权限判断、数据组装全堆在一个函数里根本没法维护。拆出 crud 层之后路由只管接收参数、调 crud、返回响应层次清爽很多。2. 数据库连接与会话管理最容易踩坑的地方要说 FastAPI 数据操作里哪部分最常见问题会话管理绝对排第一。我曾经遇到一个项目上线第二天数据库连接池就被打满了报错信息全是TimeoutError: QueuePool limit of size 10 overflow 10 reached当时排查了一晚上才发现是会话对象没有被正确关闭。2.1 引擎与会话工厂的配置细节新建一个core/database.py核心代码如下from sqlalchemy import create_engine from sqlalchemy.orm import DeclarativeBase, sessionmaker DATABASE_URL mysqlpymysql://root:passwordlocalhost:3306/fastapi_demo?charsetutf8mb4 engine create_engine( DATABASE_URL, pool_size10, max_overflow20, pool_pre_pingTrue, pool_recycle3600, echoFalse, ) SessionLocal sessionmaker(bindengine, autoflushFalse, autocommitFalse) class Base(DeclarativeBase): pass这里有几个参数我想重点讲一下因为它们直接影响到数据库的稳定性pool_size和max_overflow控制连接池的基数和溢出上限。pool_size10表示池里常驻 10 个连接max_overflow20表示高峰期最多额外再开 20 个连接也就是最大 30 个。如果应用要承受高并发这两个值要根据数据库的最大连接数来调而不是越大越好。我曾经见过有人为了性能把max_overflow设为 200结果数据库直接被连接打挂了。pool_pre_pingTrue是比较推荐开启的选项每次从池里拿连接前先发一个轻量探测确认连接没死。因为 MySQL 默认wait_timeout是 8 小时如果池里的连接空闲太久被服务端断开而客户端不知道就会拿到一个已失效的连接报Lost connection to MySQL server during query。开启这个选项能避免绝大多数这种问题。pool_recycle3600是告诉 SQLAlchemy 连接使用超过 3600 秒后要回收重建这也是为了绕开数据库服务端断连的问题双保险。echoFalse不建议在生产环境打开但开发调试阶段可以设为 True把 SQL 语句打印出来非常有助于排查问题。2.2 用依赖注入统一管理会话FastAPI 最舒服的地方在于依赖注入可以用来管理数据库会话的生命周期。你不需要在每个接口里手动SessionLocal()然后try/finally关闭只需要定义一个依赖函数from fastapi import Depends from sqlalchemy.orm import Session from app.core.database import SessionLocal def get_db(): db SessionLocal() try: yield db finally: db.close()然后在路由里这样用from fastapi import APIRouter, Depends from sqlalchemy.orm import Session router APIRouter() router.get(/users/{user_id}) def get_user(user_id: int, db: Session Depends(get_db)): return db.get(User, user_id)看到这里你可能会问为什么用yield而不是直接return db这是因为 FastAPI 的依赖系统对yield有特殊处理yield之后的代码会在请求结束时执行。于是db.close()保证了一定会被调用即使接口中途抛异常也会走finally分支。如果改成return db那么会话对象就会一直得不到释放直到被垃圾回收这在高并发下几乎必然导致连接泄漏。这里有一个小细节要提醒你autoflushFalse。这个参数关闭了自动刷新的特性防止在查询时 SQLAlchemy 先把之前没有提交的修改偷偷写进数据库导致一些很隐蔽的问题。所以事务的控制你就老老实实用commit()和rollback()来做不要依赖自动刷新。3. 模型定义与关系映射细节决定成败模型定义看起来简单不就是类加字段吗但实际上这里面的细节能卡住你半天。我见过不少把关联关系写错、把字段类型选错、把索引漏了的项目全都是前期图省事埋下的雷。3.1 字段类型与主键策略的选择先给一个比较完整的用户模型示例这里是 SQLAlchemy 2.0 的风格from sqlalchemy import String, Integer, DateTime, Boolean from sqlalchemy.orm import Mapped, mapped_column from datetime import datetime from app.core.database import Base class User(Base): __tablename__ users id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue) username: Mapped[str] mapped_column(String(50), uniqueTrue, indexTrue, nullableFalse) email: Mapped[str] mapped_column(String(120), indexTrue, nullableFalse) hashed_password: Mapped[str] mapped_column(String(255), nullableFalse) is_active: Mapped[bool] mapped_column(Boolean, defaultTrue) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow)主键我一般不搞 UUID而是用自增整数。原因也简单性能好、索引小、对大部分业务足够。UUID 虽然看起来很“分布式”但在普通单库项目里除了增加存储和索引开销意义不大。真正需要 UUID 的场景一般是多库合并、数据同步这种普通 CRUD 项目真的用不上。username这里加了uniqueTrue和indexTrue注意两者并不完全等价unique也会隐式创建索引但如果你只想要快速查询而不想限制唯一性那就只用indexTrue。所以在 username 上需要唯一约束明确写unique即可index写不写都行。created_at默认值用datetime.utcnow而不是datetime.now是避免时区混乱的常见做法后端统一存 UTC展示时再转本地时区。3.2 一对多关系外键写在“多”的那一边在文章系统里一个用户有多篇文章这就是典型的一对多关系。SQLAlchemy 2.0 里可以这样定义from typing import List from sqlalchemy import ForeignKey from sqlalchemy.orm import relationship class Article(Base): __tablename__ articles id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue) title: Mapped[str] mapped_column(String(100), nullableFalse) content: Mapped[str] mapped_column(Text, nullableFalse) author_id: Mapped[int] mapped_column(ForeignKey(users.id), indexTrue) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) author: Mapped[User] relationship(back_populatesarticles)对应在 User 模型里加articles: Mapped[List[Article]] relationship(back_populatesauthor)这里有几个容易犯的错。第一个外键字段author_id用ForeignKey(users.id)表名必须是小写复数那个表名不是类名。如果你写成ForeignKey(User.id)运行时会直接报错找不到表。第二个relationship的back_populates要成对写。我见过有人一端写了relationship另一端没写结果加数据时关联关系不生效白白排查很久。那么什么时候会用到多对多关系呢比如文章打标签一篇文章可以有多个标签一个标签也可以对应多篇文章。这种情况需要一张中间表。SQLAlchemy 里可以这样写from sqlalchemy import Table, Column article_tag Table( article_tags, Base.metadata, Column(article_id, ForeignKey(articles.id), primary_keyTrue), Column(tag_id, ForeignKey(tags.id), primary_keyTrue), ) class Tag(Base): __tablename__ tags id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue) name: Mapped[str] mapped_column(String(30), uniqueTrue, nullableFalse)多对多的relationship写起来也简单但要注意别太依赖它做复杂查询中间表上有额外的自定义字段比如“打标时间”时你其实需要的是“关联模型”而不是简单的中间表那就要用association object模式。这一篇先不展开但你遇到这种情况时要能想到这个知识点。3.3 表结构怎么创建别在生产环境直接建表很多人喜欢用Base.metadata.create_all(engine)来建表开发阶段确实简单但是你要知道它只能建表不能改表。后来你给表加了一个字段create_all不会帮你加你得把那列删了重新建。这在开发环境无所谓生产环境就是一场灾难数据全没了。所以更稳妥的方案是 Alembic。初始化方式很简单alembic init alembic然后改alembic/env.py把target_metadata指向你的Base.metadata。之后每次模型变更执行alembic revision --autogenerate -m add phone column in users alembic upgrade headautogenerate会对比模型定义和当前数据库结构自动生成迁移脚本。注意它也不是万能的有的操作比如修改字段类型、加约束它可能猜不准生成的迁移脚本一定要人工检查确认无误再执行。我个人在整个开发周期里都是开发环境随手create_all建表一旦涉及“字段变更”立刻转用 Alembic养成习惯后面对生产环境一点也不慌。4. CRUD 接口实操从最基础到常用写法这一节是大多数人最关心的部分到底怎么把增删改查写出来。我会按新增、查询、更新、删除四个场景把关键代码和背后的逻辑都过一遍。4.1 新增数据注意字段校验和事务边界新增用户接口可以这样写from fastapi import Depends, HTTPException from pydantic import BaseModel class UserCreate(BaseModel): username: str email: str password: str router.post(/users, status_code201) def create_user(payload: UserCreate, db: Session Depends(get_db)): existing db.query(User).filter(User.username payload.username).first() if existing: raise HTTPException(status_code400, detail用户名已存在) user User( usernamepayload.username, emailpayload.email, hashed_passwordpayload.password, # 实际应该做哈希这里简化 ) db.add(user) db.commit() db.refresh(user) return user我印象里最常被问到的点是为什么db.add()之后还要db.commit()和db.refresh()这里展开一下db.add()只是把对象放入会话的待处理队列此时数据库里还没有记录db.commit()把事务提交数据真正落库同时 SQLAlchemy 会使用数据库生成的自增 ID 回填到user.iddb.refresh()是重新从数据库加载一遍这个对象确保返回给前端的数据是最新的特别是我们有数据库默认值字段的时候比如created_at。如果我不想要refresh()也可以只commit()然后实例的id通常也已经有了但created_at这类由数据库默认值生成的字段可能没刷新保险起见还是写refresh。还有一件事上面代码为了演示简洁密码是明文存的实际项目里一定要用 hash。这一篇主聊数据操作但请一定记住生产项目里密码必须哈希处理。4.2 查询数据get 还是 filter优先级问题查询单个对象有两个常用方式user db.get(User, user_id) # 推荐 user db.query(User).filter(User.id user_id).first() # 经典写法db.get()是 SQLAlchemy 2.0 推荐的用法直接按主键查询它有一个好处如果会话里已经有了这个对象比如刚新增过它可能直接从会话缓存返回省一次数据库访问。db.query()则是老版本一直流传下来的写法虽然 SQLAlchemy 2.0 里没有删除它但官方现在推荐用select()这种新风格。两种都可以用关键是要在一个项目里保持一致不要今天写一种明天写一种。列表查询最常见的需求是按条件过滤、按时间排序、分页。我会这样写router.get(/users) def list_users( skip: int 0, limit: int 20, keyword: str None, db: Session Depends(get_db), ): query db.query(User).order_by(User.created_at.desc()) if keyword: query query.filter(User.username.contains(keyword)) users query.offset(skip).limit(limit).all() return users这里有一个挺重要的细节filter的条件是User.username.contains(keyword)它在 MySQL 里对应的是LIKE %keyword%这种写法不会走索引当用户表数据量大时查询会慢。你可以在条件里换用startswithLIKE keyword%这种前缀匹配在数据库里能用到索引性能好很多。4.3 更新和删除先取后改别省略查询更新操作的常规写法router.put(/users/{user_id}) def update_user(user_id: int, payload: UserUpdate, db: Session Depends(get_db)): user db.get(User, user_id) if not user: raise HTTPException(status_code404, detail用户不存在) if payload.username is not None: user.username payload.username if payload.email is not None: user.email payload.email db.commit() db.refresh(user) return user这里要特别注意一个点更新前先查一次对象除了要检查是否存在之外也是为了避免“盲目更新”。比如你直接db.query(User).filter(User.id user_id).update({...})看起来效率很高但问题在于如果 id 不存在你也不知道还得再查一次你无法在更新前做一些字段级校验比如 email 格式、用户名是否重复不会触发 SQLAlchemy 的 ORM 事件钩子比如更新时间的自动处理。删除操作也有一个常见误区。很多人会在删除前省略查询直接db.query(User).filter(User.id user_id).delete()然后 commit最后返回 200哪怕数据库里根本没有这一行。更稳的做法还是先查一次router.delete(/users/{user_id}) def delete_user(user_id: int, db: Session Depends(get_db)): user db.get(User, user_id) if not user: raise HTTPException(status_code404, detail用户不存在) db.delete(user) db.commit() return {ok: True}我自己在写删除接口时还会考虑一个问题这个删除是“物理删除”还是“逻辑删除”对于核心业务数据比如订单、用户直接物理删除一旦误操作很难恢复。所以不少项目会给表加一个is_deleted字段查询时默认过滤掉已删除的记录这就是所谓的软删除。如果你们业务上有这个需求建议从第一版就把这个字段加上不然后面加软删除要改的查询非常非常多。5. 分页、过滤与查询性能列表接口的必修课列表接口是每个项目都离不开的。但你有没有发现很多新手写的列表接口数据量一上来就卡成狗这节我们专门讲分页和常见的性能问题。5.1 分页offset 分页还是 cursor 分页FastAPI 的列表接口最经典的分页方案是router.get(/articles) def list_articles( page: int 1, page_size: int 10, db: Session Depends(get_db), ): offset (page - 1) * page_size total db.query(Article).count() items db.query(Article).order_by(Article.created_at.desc()).offset(offset).limit(page_size).all() return {total: total, items: items, page: page, page_size: page_size}这种基于OFFSET的分页在数据量小于几万的时候完全够用写起来也简单前端拿total做分页组件非常方便。但当你数据量超过几十万、上百万以后offset 分页会越来越慢因为数据库每次都要跳过前 N 条记录越到后面越吃力。如果你的列表页要支持无限滚动或者数据量真的很大可以改成基于游标的分页。最简单的实现是用主键 id 作为游标router.get(/articles) def list_articles( cursor_id: int 0, page_size: int 10, db: Session Depends(get_db), ): items ( db.query(Article) .filter(Article.id cursor_id) .order_by(Article.id.asc()) .limit(page_size) .all() ) next_cursor items[-1].id if len(items) page_size else None return {items: items, next_cursor: next_cursor}这种方式的查询条件走主键索引速度极快而且不受页数影响。代价是前端拿不到total也无法随意跳转到某一页。它适合“往下刷”的信息流场景不适合后台管理系统。所以你选分页方式前先搞清楚业务场景。5.2 N1 查询问题看似简单实则致命N1 查询是 ORM 性能问题里最常出现的举一个典型例子查询文章列表时每篇文章要显示作者名。如果这样写articles db.query(Article).limit(20).all() for article in articles: print(article.author.username) # 每次访问都会额外发一条 SQL 查用户第一次查询返回 20 篇文章然后每篇文章访问.author时都会再发一条 SQL 查用户总共执行 1 20 条 SQL。这就是 N1。当列表页文章很多时数据库会被这些小查询拖垮。解决办法是预加载关联对象。在 SQLAlchemy 2.0 里使用selectinloadfrom sqlalchemy.orm import selectinload articles ( db.query(Article) .options(selectinload(Article.author)) .limit(20) .all() )这样 SQLAlchemy 会先查出 20 篇文章再用一条WHERE id IN (...)查出对应的作者SQL 总数从 21 条降到 2 条。还有一个古老的joinedload它用 JOIN 实现但遇到集合类关系时容易把结果集撑大在列表场景下我更推荐selectinload简洁且不容易出错。5.3 一个列表接口的性能优化清单如果你的列表接口慢可以按这个顺序排查我基本每次都这样做第一看 SQL 是否走了索引。在 MySQL 里用EXPLAIN SELECT ...检查type列如果是ALL说明在做全表扫描这时候要么加索引要么调整查询条件。第二看是否查了不必要的字段。返回列表时如果只需要id, title, created_at这些字段不要用SELECT *。可以借助with_entities只查询需要的列减少数据传输和 ORM 对象构造的开销。第三看关联关系有没有触发 N1。打开 SQL 日志观察一下如果同一个查询反复出现多次十有八九是没加预加载。第四看分页怎么写的。如果数据量大又用了深分页比如OFFSET 100000看看能不能改成游标分页。第五真的到了数据量极大、复杂查询很多的地步考虑引入 Redis 做缓存或者用专门的搜索服务但普通项目通常走到前四步就够了。6. 常见问题排查与避坑记录最后这部分我把自己和数据操作打过交道的典型问题梳理一遍都是真实踩过的坑每一类问题都给出排查思路和预防方案。6.1 数据库连接池耗尽接口全部卡死现象接口突然大面积超时日志里出现QueuePool limit reached之类的报错或者 MySQL 端显示连接数打满。排查思路先确认是不是代码里到处手动创建 Session 而忘记关闭。尤其在异步任务、定时任务里最容易犯这个错。我还在一个项目里遇到过有人把SessionLocal()写在了模块顶层等于全局就一个会话被所有人都用这不只是连接池问题并发下数据错乱都是可能的。预防与解决所有 Session 的创建和关闭都要成对出现能走 FastAPI 依赖注入就尽量别手动管理。另外给连接池设置一个合理的上限不要无脑开大。临时救急时重启应用可以释放连接但根源是代码必须改对。6.2 修改数据后查询结果还是旧的现象接口更新了数据但下一次 GET 返回的还是旧值或者列表里新加的记录查不到。这类问题很大概率是事务隔离问题。MySQL 默认是REPEATABLE READ长事务里读取的是事务开始时的快照。如果你在同一个会话里既做了更新又做了查询还迟迟不 commit那其他会话可能看不到你的修改。更常见的情况是FastAPI 的依赖用了yield管理会话但你在接口里 commit 之后又做了一次查询而 SQLAlchemy 的会话缓存还持有旧对象就会返回旧值。这时可以用db.expire_all()让会话下次访问时强制刷新或者直接重新db.get()。生产环境里我建议事务提交后把相关对象db.refresh()一下别偷懒。6.3 删除关联数据时外键报错现象删除一条用户数据时报ForeignKeyViolation因为它下面还有文章记录。这个本质上不是 FastAPI 的问题是表设计时外键约束策略没想好。SQLAlchemy 模型里可以控制外键的ondelete行为author_id: Mapped[int] mapped_column( ForeignKey(users.id, ondeleteCASCADE), indexTrue )CASCADE表示删用户时自动删文章SET NULL表示把文章的作者置空前提是字段允许为空。到底选哪种取决于业务逻辑。比如文章被删了作者很多业务是不能接受的那你更适合用软删除或者SET NULL。我这里再提醒一句外键约束设置的ondelete要跟数据库实际行为一致如果你依赖 ORM 删除SQLAlchemy 也会尝试遵守这个约束但不一致时会报错。6.4 接口返回 JSON 时序列化失败现象GET /users/1返回 500日志里提示Object of type User is not JSON serializable。很多人第一次遇到都懵明明 User 是普通对象为什么不能序列化因为 FastAPI 默认的 JSON 编码器只能处理 dict、list、基础类型而 SQLAlchemy 模型是 ORM 对象。解决方式有两种第一种在响应模型里声明 Pydantic schema让 FastAPI 自动做from_attributesTrue转换。这种方式代码整洁强烈推荐。class UserOut(BaseModel): id: int username: str email: str model_config {from_attributes: True}然后在路由里写response_modelUserOutFastAPI 就能把一个 SQLAlchemy 对象按 schema 的字段序列化。第二种手动转字典。用user.__dict__或者写一个 to_dict 方法。这种方式简单粗暴但字段控制不灵活临时调试可以长期不建议。6.5 开发调试时如何快速看 SQL排查查询性能问题时最直接的帮手就是把执行的 SQL 打出来。除了在create_engine里设置echoTrue之外我更喜欢用 SQLAlchemy 的事件监听只打印SELECT语句避免刷屏from sqlalchemy import event from sqlalchemy.engine import Engine import logging logger logging.getLogger(sqlalchemy.engine.Engine) event.listens_for(engine, before_cursor_execute) def _before_execute(conn, cursor, statement, parameters, context, executemany): if statement.lstrip().upper().startswith(SELECT): logger.info(SQL: %s, statement)在调试 N1 问题时这个打印太有用了只要看到某个查询反复出现马上就能定位到问题。聊了这么多其实数据操作这块真正重要的不是记住某一条代码而是形成一套流程先设计好目录和模型再管好会话生命周期然后写 CRUD 时保持一致性最后在列表接口上多留个心眼优化性能。我自己早期写 FastAPI 项目时最深的感触就是数据层的问题往往不会立刻暴露都是一点点积累到线上才爆出来。希望这篇能帮你把这些坑提前绕过去后面几篇我们继续聊认证、中间件和部署。