终于写到第5部分了。前四部分我们把FastAPI的路由、参数校验、请求体、响应模型、中间件和异常处理都过了一遍单看接口已经能跑得像模像样。但做过真实项目的人都知道这还只是“玩具级”的后端。原因很简单数据一重启就丢接口之间没法共享状态更别提多实例部署时的数据一致性了。所以第5部分的核心任务就是给FastAPI接上真正的数据库用SQLAlchemy ORM做持久化把内存里的假数据搬到关系型数据库里顺便把CRUD接口从“能用”打磨成“抗造”。这部分内容适用于所有准备把FastAPI项目推向实际的开发者不管你是学生项目、个人作品还是内部工具看完就能直接抄作业。1. 前置思路为什么第5部分要讲数据库集成1.1 从内存数据到持久化存储的痛点前面几部分的Demo我猜大部分人都和我一样数据直接用Python字典或者列表存着接口里读写内存。好处是零配置、即时可用坏处却很致命进程重启等于数据清零多个工作进程各持一份内存副本运维想做个滚动更新都提心吊胆。这时候你才意识到后端应用真正值钱的部分不是那一层路由转发而是数据资产。数据库的作用就是提供一份“所有实例共享的、可以长期落盘的数据源”让业务逻辑和存储逻辑彻底解耦。从FastAPI角度切入它本身不绑定任何数据库但通过依赖注入机制你可以非常优雅地把数据库会话注入到每个接口函数里。FastAPI的高明之处在于它的依赖系统是原生的、带缓存和生命周期管理的所以数据库连接的开闭、事务的提交回滚都能被安排得清清楚楚。配合Pydantic做数据校验再配合SQLAlchemy做ORM映射三层结构基本就成了Python后端的黄金组合。1.2 技术选型SQLAlchemy ORM PostgreSQL/SQLite以及为什么选它先说为什么选SQLAlchemy。Python的数据库生态里有原生sqlite3、有Django ORM、有Tortoise ORM、有Peewee但SQLAlchemy是当之无愧的行业标准。它既有Core的SQL表达式语言也有高层的ORM层你既可以写查询对象也可以在必要时落到原生SQL灵活性是其他框架比不了的。更关键是它的会话Session管理机制配合FastAPI的依赖注入能天然形成“请求进来开会话、请求结束关会话”的边界。数据库本身开发阶段我用SQLite零配置、单文件方便极了生产阶段换PostgreSQL成熟稳定、支持并发写入、支持JSONB、全文检索这些实用特性。大家不用纠结选型因为SQLAlchemy的方言层已经帮我们把大部分差异屏蔽了切换数据库通常只需要改一个连接字符串。但要注意千万别把开发和生产用完全不同的数据库而忽略方言差异比如SQLite不支持ALTER TABLE这种常见的迁移坑这是后话。2. 环境准备与项目结构拆分2.1 安装依赖与标准目录在写代码之前先把依赖装齐。这里给出一份适合FastAPI SQLAlchemy项目的requirements清单fastapi0.115.0 uvicorn[standard]0.30.0 sqlalchemy2.0.30 pydantic2.7.0 pydantic-settings2.3.0 psycopg2-binary2.9.9 alembic1.13.1 python-multipart0.0.9sqlalchemy和pydantic是核心alembic是后面做迁移用的psycopg2-binary是PostgreSQL驱动如果你只用SQLite可以暂时不装。Python版本我建议3.10以上因为新版SQLAlchemy和Pydantic都越来越依赖新语法老版本会遇到各种兼容性小毛病。目录结构建议从一开始就按“可扩展”的方式组织而不是把所有东西堆在main.py里。我自己的惯用结构是这样的app/ ├── __init__.py ├── main.py # FastAPI实例、路由挂载 ├── config.py # 配置项数据库URL等 ├── database.py # engine、SessionLocal、Base ├── models/ # SQLAlchemy ORM模型 │ ├── __init__.py │ └── item.py │ └── user.py ├── schemas/ # Pydantic模型输入/输出 │ ├── __init__.py │ └── item.py │ └── user.py └── routers/ # APIRouter ├── __init__.py └── items.py └── users.py可能有同学觉得分这么细小题大做但项目一旦超过几个接口模型、schema、路由混在一起就非常痛苦。第5部分我们允许自己养成这个好习惯。2.2 数据库连接配置与URL写法数据库连接字符串是整个数据层的“地基”。SQLite的URL很简单DATABASE_URL sqlite:///./fastapi_demo.dbPostgreSQL长这样DATABASE_URL postgresqlpsycopg2://user:passwordlocalhost:5432/myapp我习惯把这些配置放在config.py里用pydantic-settings来读取环境变量而不是硬编码在代码里from pydantic_settings import BaseSettings class Settings(BaseSettings): database_url: str sqlite:///./fastapi_demo.db echo_sql: bool False class Config: env_prefix APP_ settings Settings()这样你运行export APP_DATABASE_URLpostgresql...就能覆盖默认值不用改代码。database.py里再统一初始化引擎from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, declarative_base from config import settings engine create_engine( settings.database_url, connect_args{check_same_thread: False} if settings.database_url.startswith(sqlite) else {}, echosettings.echo_sql, ) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) Base declarative_base()SQLite那个check_same_thread: False参数很关键因为FastAPI默认在多线程下处理请求SQLite默认只允许创建它的线程访问连接加了这个参数才能跨线程使用。如果是PostgreSQL就不用管。2.3 模型与表的设计Column、关系模型层是ORM的核心。我们来设计两个实体分类Category和物品Item顺便演示一对多关系。先看分类模型from sqlalchemy import Column, Integer, String from database import Base class Category(Base): __tablename__ categories id Column(Integer, primary_keyTrue, indexTrue) name Column(String(50), nullableFalse, uniqueTrue)物品模型from sqlalchemy import Column, Integer, String, ForeignKey from sqlalchemy.orm import relationship from database import Base class Item(Base): __tablename__ items id Column(Integer, primary_keyTrue, indexTrue) name Column(String(100), nullableFalse) price Column(Integer, nullableFalse) # 单位分避免浮点误差 category_id Column(Integer, ForeignKey(categories.id), nullableFalse) category relationship(Category, back_populatesitems)注意我刻意用了nullableFalse和合理长度的字符串这能帮数据库拦住大部分非法数据。价格用整数分存储而不是浮点数这是做支付类业务的老经验浮点数精度坑太多了。关系这里容易犯的错只在“多”的一方定义外键但忘了在“一”的一方加反向引用。给Category补上items relationship(Item, back_populatescategory)这样你就能通过category.items反向拿到物品列表了。ORM关系不会自动帮你在数据库层面建外键约束但SQLAlchemy在create_all创建表时会把ForeignKey转为真正的约束所以这种声明已经够了。3. 核心实操从模型到CRUD接口3.1 用Pydantic模式与ORM模型的解耦Schema这是FastAPI初学者最容易搞混的一层。模型ORM Model是给数据库用的Pydantic Schema是给API请求和响应用的。两者不要混写否则会出现“响应里多返回了password”“客户端传了不该传的字段”之类的隐患。以Item为例我们应该定义三套schemafrom pydantic import BaseModel, ConfigDict class ItemBase(BaseModel): name: str price: int class ItemCreate(ItemBase): category_id: int class ItemUpdate(BaseModel): name: str | None None price: int | None None category_id: int | None None class ItemOut(BaseModel): model_config ConfigDict(from_attributesTrue) id: int name: str price: int category_id: int关键就在ItemOut的model_config ConfigDict(from_attributesTrue)这行代码告诉Pydantic可以从ORM对象的属性读取数据否则你直接把ORM对象放到response_model里会报校验错误。Pydantic v2的写法是from_attributesTrue老项目里写的是orm_modeTrue如果你是照着旧教程抄的记得改成新写法。3.2 实现依赖注入get_db会话管理FastAPI依赖注入的经典场景就是数据库会话。先定义生成器依赖from database import SessionLocal def get_db(): db SessionLocal() try: yield db finally: db.close()这个依赖一定要用yield不能直接return db。因为FastAPI只有在请求结束、响应返回给客户端之后才会执行yield后面的清理代码。用try/finally保证无论接口是否抛异常会话都会关闭避免连接泄漏。一旦连接泄漏PostgreSQL的连接池会被慢慢占满线上表现就是“突然接口全部卡死”。路由里通过参数注入dbfrom fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from models.item import Item from schemas.item import ItemCreate, ItemOut router APIRouter(prefix/items, tags[items]) router.post(/, response_modelItemOut, status_code201) def create_item(item: ItemCreate, db: Session Depends(get_db)): db_item Item(**item.model_dump()) db.add(db_item) db.commit() db.refresh(db_item) return db_item注意几个动作model_dump()把Pydantic对象转字典然后展开传给ORM构造器。commit()之后ORM对象已经有了主键但还没有数据库生成的默认值所以用refresh()重新加载一次保证响应数据完整准确。3.3 CRUD接口实现与状态码处理一个完整的Items接口应该覆盖增删改查。我建议把同样的逻辑扩展到分类接口这里给出所有核心端点。列表查询要支持简单的分页router.get(/{item_id}, response_modelItemOut) def get_item(item_id: int, db: Session Depends(get_db)): item db.get(Item, item_id) if not item: raise HTTPException(status_code404, detailItem not found) return item router.get(/, response_modellist[ItemOut]) def list_items(skip: int 0, limit: int 20, db: Session Depends(get_db)): return db.query(Item).offset(skip).limit(limit).all()分页参数用skip/limit简单直接适合内部系统。如果是C端产品建议改成基于游标的分页不过那是后面的主题了。更新接口更值得留意router.patch(/{item_id}, response_modelItemOut) def update_item(item_id: int, payload: ItemUpdate, db: Session Depends(get_db)): item db.get(Item, item_id) if not item: raise HTTPException(status_code404, detailItem not found) update_data payload.model_dump(exclude_unsetTrue) for field, value in update_data.items(): setattr(item, field, value) db.commit() db.refresh(item) return item这里最容易被忽略的是exclude_unsetTrue只有传了参数的字段才更新否则前端只改了个名称你把price也覆盖成None了那就踩大坑。删除接口同理router.delete(/{item_id}, status_code204) def delete_item(item_id: int, db: Session Depends(get_db)): item db.get(Item, item_id) if not item: raise HTTPException(status_code404, detailItem not found) db.delete(item) db.commit() return None删除成功返回204无响应体这是RESTful的最佳实践强迫自己不再为了好看而返回数据。3.4 关系字段的序列化嵌套模型的坑前面我们设计了Category和Item的一对多关系实际做接口时怎么返回嵌套结构最常见需求是“返回分类时把该分类下的物品列表一起带上”。在Pydantic Schema里可以这么定义class CategoryOut(BaseModel): model_config ConfigDict(from_attributesTrue) id: int name: str items: list[ItemOut] []然后分类路由返回时直接传ORM对象router.get(/categories/{category_id}, response_modelCategoryOut) def get_category(category_id: int, db: Session Depends(get_db)): category db.get(Category, category_id) if not category: raise HTTPException(status_code404, detailCategory not found) return category这个方案在数据量小的时候没问题但有一个隐藏的N1查询问题当你返回多个分类时SQLAlchemy会为每个分类单独发起一次SELECT * FROM items WHERE category_id ...100个分类就是101条SQL。性能优化可以用selectinload或joinedload预加载比如from sqlalchemy.orm import selectinload category db.query(Category).options(selectinload(Category.items)).first()很多新手都会卡在这接口能跑但特别慢其实就是没加预加载。后面我们单独说性能陷阱。4. 迁移、初始化与测试技巧4.1 建表策略Base.metadata.create_all vs Alembic很多教程直接让你在启动时调用Base.metadata.create_all(engine)这个方案适合开发初期但它不会自动修改已存在的表结构。你改了一个字段名重跑create_all是不生效的。所以到了第5部分我强烈建议引入Alembic做迁移管理。先初始化alembic init alembic然后修改alembic/env.py把target_metadata指向我们的Base.metadatafrom database import Base target_metadata Base.metadata首次迁移记录所有表结构alembic revision --autogenerate -m create tables alembic upgrade head从此以后每次修改模型后跑一下revision --autogenerate就能生成迁移脚本再upgrade head应用到数据库。注意autogenerate不是万能的它检测不到列重命名、默认值变更这类“语义性”变化生成后一定要人工检查脚本。4.2 用FastAPI依赖覆盖做测试隔离写测试的时候不能连开发库否则一遍遍跑测试会把数据搞脏。常规做法是用一台测试专用的SQLite内存库并且通过FastAPI的依赖覆盖机制把get_db换掉。代码示例from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool from database import Base from main import app from dependencies import get_db engine create_engine( sqlite://, connect_args{check_same_thread: False}, poolclassStaticPool, ) TestingSessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) Base.metadata.create_all(bindengine) def override_get_db(): db TestingSessionLocal() try: yield db finally: db.close() app.dependency_overrides[get_db] override_get_db client TestClient(app)这里必须用StaticPool因为SQLite内存库默认每个连接都是独立数据库不用StaticPool的话你在创建测试数据的连接和FastAPI拿到的连接不是同一个库查不到数据。这个小坑能浪费一晚上。跑测试时每个用例前后最好清库最简单的办法是测试函数里先删除所有记录或者用pytest fixture逐个事务回滚后者更高级一些。4.3 常见问题排查与性能陷阱我把自己踩过以及帮别人排查过的高频问题整理一下现象常见原因解决办法请求完成后session is already flushing接口里没有关闭session或者依赖没走完检查是否把get_db写成了普通函数必须用yield依赖启动后数据库表没生成忘了引入模型模块在main.py中显式import models或者import app.models查询返回慢且N条记录发出N1条SQL未预加载关系字段使用selectinload或joinedload更新时字段被置空model_dump没加exclude_unsetpayload.model_dump(exclude_unsetTrue)并发读写死锁事务时间过长缩短事务、用读提交隔离级别减少锁竞争Postgres连接数打满SessionLocal被直接当单例使用确认每个请求都新建Session而不是全局共用一个session还有一个常被忽视的点FastAPI处理Pydantic校验是在依赖注入之前如果参数非法get_db根本不会被调用数据库连接不会泄漏。但如果你的依赖里生成了数据库会话却在路由里因为异常提前返回没有走完finally那就要靠db.rollback()配合。安全做法是在抛出HTTPException前让内存对象与数据库解绑一般用db.expunge(item)或直接db.rollback()。从性能角度第一要注意的是连接池。默认pool_size是5max_overflow是10对一个小应用够用但并发一高就抓瞎。调法如下engine create_engine( settings.database_url, pool_size20, max_overflow10, pool_pre_pingTrue, )pool_pre_pingTrue我强烈建议打开它会在每次取连接之前发送一个SELECT 1检查连接是否存活。数据库重启、网络抖动都会像幽灵一样产生断开的连接不开启pre_ping你可能会看到间歇性的server closed the connection unexpectedly开了之后这个错误基本消失。另一个性能陷阱是用ORM批量更新。循环里一条条执行update会跑得很慢for item in items: item.price * 1.2 db.commit()SQLAlchemy会把所有变更缓存在内存中flush时才逐条UPDATE如果几千条数据可以先db.bulk_update_mappings()几万条干脆用SQLAlchemy Core或原生SQL批量UPDATE速度能差两个数量级。4.4 实测心得连接池与超时配置最后分享一下我实际部署时的配置。开发环境用SQLite连接参数保持默认。生产环境用PostgreSQLengine配置如下connect_args { connect_timeout: 10, keepalives: 1, keepalives_idle: 30, keepalives_interval: 10, keepalives_count: 5, } engine create_engine( settings.database_url, pool_size10, max_overflow20, pool_pre_pingTrue, pool_recycle3600, connect_argsconnect_args, )pool_recycle很关键PostgreSQL有时会主动断开闲置超过一定时间的连接如果连接池里还拿着旧连接就会报错。设成3600秒就让连接最多存活1小时强制从底层新建比等到报错再去重连舒服得多。超时方面接口层面的数据库操作本身要快但慢查询是不可避免的比如报表此时建议给语句设置超时。PostgreSQL可以通过SET statement_timeout实现SQLAlchemy语句执行前统一设置比较麻烦一个简单做法是在Session绑定的事件处理器里执行from sqlalchemy import event from sqlalchemy.engine import Engine event.listens_for(Engine, connect) def set_sqlite_pragma(dbapi_connection, connection_record): if settings.database_url.startswith(sqlite): cursor dbapi_connection.cursor() cursor.execute(PRAGMA foreign_keysON) cursor.close()这个监听器还帮我处理了SQLite外键约束默认关闭的问题。没错SQLite默认不启用外键约束如果你希望删除分类时因为还有物品而报错必须每条连接执行PRAGMA foreign_keysON。不写这行你创建的ForeignKey就形同虚设。我个人在实际操作中的体会是FastAPI接入SQLAlchemy之后项目才真正进入可交付阶段。别被“ORM很难”吓退只要把模型、Schema、依赖注入三条线理清楚大部分日常需求都能顺手搞定。刚开始写CRUD时建议多关注边界条件不存在返回404、字段校验失败返回422、删除关联数据被外键拦截返回500这些都是真实用户会碰到的情况。后面有时间可以再往认证授权、Alembic自动部署、读写分离上深入但先把这块地基打牢后面做什么都稳。