FastAPI 集成 SQL 数据库实战：基于 SQLModel 的单模型与多模型 CRUD 开发指南

FastAPI 集成 SQL 数据库实战基于 SQLModel 的单模型与多模型 CRUD 开发指南【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi本文以 FastAPI 官方教程中「SQL关系型数据库」章节为骨架讲解如何用 SQLModelSQLAlchemy 与 Pydantic 的组合为 FastAPI 应用接入 SQLite/PostgreSQL 等关系型数据库并完成从「单模型最小可用」到「多模型安全重构」的完整演进。读完本文你将掌握 SQLModel 表模型与数据模型的建模、FastAPI 依赖注入式 Session 管理、基于模型注解的自动请求校验/响应序列化以及用HeroCreate/HeroPublic/HeroUpdate等角色化模型保护 API 边界与敏感字段的工程手法。官方教程的完整代码示例位于 docs_src/sql_databases/含tutorial001_an_py310.py与tutorial002_an_py310.py配套自动化测试位于 tests/test_tutorial/test_sql_databases/均可直接查阅、运行与验证。FastAPI 并不强制你使用哪种数据库FastAPI 本身不要求你使用 SQL关系型数据库你可以按需选择任何 SQL 或 NoSQL 方案其中一些被称为 ORM——用类表示 SQL 表、用实例表示表中行的对象关系映射包。官方教程以SQLModel作为推荐示例SQLModel 建立在 SQLAlchemy 与 Pydantic 之上由 FastAPI 同一作者创建被视为 FastAPI SQL 场景的「最佳组合」由于 SQLModel 底层就是 SQLAlchemy因此SQLAlchemy 支持的数据库 SQLModel 全部支持包括 PostgreSQL、MySQL、SQLite、Oracle、Microsoft SQL Server 等教程示例选用SQLite因为它只是单个文件且 Python 内置支持复制即可直接运行而生产环境通常建议切换到PostgreSQL这类独立数据库服务器。从该仓库的 pyproject.toml 可确认docs_src/sql_databases/下的示例直接依赖sqlmodel、sqlalchemy测试中还出现StaticPool这些依赖共同支撑着本教程的完整可运行性。安装 SQLModel在项目根目录执行$ uv add sqlmodel --- 100%第一阶段单模型最小应用先构建最简单版本只用一个 SQLModel 模型完成建表与增删查改。完整代码见 tutorial001_an_py310.py另有不使用Annotated的 tutorial001_py310.py 变体。创建表模型Table Modelfrom typing import Annotated from fastapi import Depends, FastAPI, HTTPException, Query from sqlmodel import Field, Session, SQLModel, create_engine, select class Hero(SQLModel, tableTrue): id: int | None Field(defaultNone, primary_keyTrue) name: str Field(indexTrue) age: int | None Field(defaultNone, indexTrue) secret_name: strHero看起来很像一个普通 Pydantic 模型——事实上它底层就是一个 Pydantic 模型。但它有三点关键差异写法含义tableTrue告诉 SQLModel 这是一个表模型对应数据库里的一张表不带该参数的是普通数据模型类似纯 Pydantic 类Field(primary_keyTrue)把id声明为 SQL主键Field(indexTrue)为该列创建 SQL索引按此列过滤的查询会更快关于主键有几个值得注意的细节代码中把主键写成id: int | None是为了能在 Python 侧先创建不带id的对象idNone把生成id的工作交给数据库SQLModel 会理解这一点并在建表 schema 中把该列定义为非空的 INTEGER声明为str的字段SQLModel 会映射为 SQL 的TEXT或视数据库类型为VARCHAR。创建 Engine维持数据库连接sqlite_file_name database.db sqlite_url fsqlite:///{sqlite_file_name} connect_args {check_same_thread: False} engine create_engine(sqlite_url, connect_argsconnect_args)一个 SQLModelengine底层是 SQLAlchemy engine负责维持到数据库的连接整个代码库连接同一数据库时只应有一个engine单例对象。其中check_same_threadFalse允许 FastAPI 在不同线程中使用同一个 SQLite 数据库——这一点是必要的因为一次请求可能跨多个线程例如在依赖解析过程中。后续代码会通过「每个请求一个 Session」的结构真正贯彻这一意图。创建数据表def create_db_and_tables(): SQLModel.metadata.create_all(engine)该函数通过SQLModel.metadata.create_all(engine)为所有表模型创建对应数据表。用依赖注入管理 Session每请求一个 Sessiondef get_session(): with Session(engine) as session: yield session SessionDep Annotated[Session, Depends(get_session)]Session负责在内存中保存对象、跟踪数据变化再借助engine与数据库通信。通过 FastAPI 的yield依赖为每个请求提供一个新的Session并在请求结束后随with上下文自动关闭这正是「每请求一个 Session」的实现方式。随后定义的类型别名SessionDepAnnotated[Session, Depends(get_session)]让后续代码书写更简洁。启动时建表app FastAPI() app.on_event(startup) def on_startup(): create_db_and_tables()在应用启动事件里创建数据表。对生产环境更稳妥的做法是使用在应用启动前执行的迁移脚本。仓库测试中对教程模块的加载注释写着# TODO: remove when updating SQL tutorial to use new lifespan API提示该示例目前仍使用app.on_event(startup)写法未来教程可能迁移到新版 lifespan 机制——了解这一点有助于你衔接新旧两种 FastAPI 生命周期写法。提示SQLModel 未来会提供基于 Alembic 的迁移工具封装在那之前可以先用 Alembic 直接做迁移。创建 Hero把表模型当请求体与响应体因为每个 SQLModel 模型同时也是 Pydantic 模型所以它能直接用在各种类型注解的位置把参数类型声明为HeroFastAPI 就会从JSON body读取并校验把它作为函数返回类型注解其数据形态就会出现在自动 API 文档、并用于响应序列化。app.post(/heroes/) def create_hero(hero: Hero, session: SessionDep) - Hero: session.add(hero) session.commit() session.refresh(hero) return hero流程是把新Hero加入 Session →commit()提交到数据库 →refresh()刷新对象拿到数据库生成的id→ 返回。读取 Heroes分页查询app.get(/heroes/) def read_heroes( session: SessionDep, offset: int 0, limit: Annotated[int, Query(le100)] 100, ) - list[Hero]: heroes session.exec(select(Hero).offset(offset).limit(limit)).all() return heroes用select(Hero)构造查询并配合offset/limit做分页其中limit通过Query(le100)限制了最大取值最多 100。测试中client.get(/heroes/?offset1limit1)返回单条记录的断言正是对分页行为的验证见 test_tutorial001.py。读取单个 Heroapp.get(/heroes/{hero_id}) def read_hero(hero_id: int, session: SessionDep) - Hero: hero session.get(Hero, hero_id) if not hero: raise HTTPException(status_code404, detailHero not found) return hero按主键用session.get(Hero, hero_id)读取找不到时抛出404。配套测试同时覆盖了「删除/更新不存在的 hero 返回 404」的边界情况。删除一个 Heroapp.delete(/heroes/{hero_id}) def delete_hero(hero_id: int, session: SessionDep): hero session.get(Hero, hero_id) if not hero: raise HTTPException(status_code404, detailHero not found) session.delete(hero) session.commit() return {ok: True}运行第一阶段应用$ uv run fastapi dev INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRLC to quit)打开/docs交互式文档可以看到 FastAPI 正利用这个Hero模型同时完成 API 文档生成、数据校验与响应序列化。第二阶段多模型重构提升安全与灵活性单模型版本存在两个安全隐患原教程称之为 级问题客户端可以提交idAPI 把整个Hero当请求体接收等于允许调用方决定id甚至覆盖数据库中已存在的记录。决定id应是后端/数据库的职责而不是客户端secret_name被原样返回到处把秘密身份暴露给客户端就谈不上「secret」了。解决办法是引入多个角色化模型。这正是 SQLModel 的高光之处tableTrue的类是表模型没有tableTrue的类就是数据模型本质是 Pydantic 模型带少量扩展能力通过类继承可以避免在多个模型中重复声明字段。重构后的完整代码见 tutorial002_an_py310.py。HeroBase公共字段基类先定义容纳所有模型共享字段name、age的基类class HeroBase(SQLModel): name: str Field(indexTrue) age: int | None Field(defaultNone, indexTrue)Hero真正的表模型表模型Hero额外持有并非每个模型都出现的字段id与secret_name。由于继承了HeroBaseHero的完整字段为id、name、age、secret_name。class Hero(HeroBase, tableTrue): id: int | None Field(defaultNone, primary_keyTrue) secret_name: strHeroPublic返回给客户端的公开模型HeroPublic用于返回给 API 客户端它与HeroBase字段一致因此不会包含secret_name秘密身份终于得到保护 。它重新声明了id: int——这是与 API 客户端签订的契约id始终存在且是int永不可能是None。class HeroPublic(HeroBase): id: int对客户端而言「返回字段一定存在且类型确定」让调用代码可以写得更简单对自动生成的客户端来说也会得到更简洁的接口定义。HeroCreate用于创建的数据模型HeroCreate用于校验客户端提交的数据包含HeroBase的全部字段另加secret_name。这样客户端创建英雄时可提交secret_name入库但它不会出现在响应里。class HeroCreate(HeroBase): secret_name: str这正是处理密码类敏感数据的标准姿势接收它、绝不回传它更进一步存储前应先哈希处理永远不要明文入库。HeroUpdate用于更新的数据模型单模型版本缺少「更新英雄」能力多模型时代补上了。HeroUpdate特殊之处在于字段与创建所需字段相同但每个字段都可选带默认值这样更新时只需发送想改的字段。class HeroUpdate(HeroBase): name: str | None None age: int | None None secret_name: str | None None由于字段类型确实变化了类型包含None、默认值也是None所以必须逐个重新声明——其实没有必要继承HeroBase教程中继承它纯粹出于代码一致性/个人偏好。创建收 HeroCreate返 HeroPublicapp.post(/heroes/, response_modelHeroPublic) def create_hero(hero: HeroCreate, session: SessionDep): db_hero Hero.model_validate(hero) session.add(db_hero) session.commit() session.refresh(db_hero) return db_hero请求收到HeroCreate由此构造表模型Hero通过Hero.model_validate(hero)完成转换入库后由数据库生成id。函数虽返回表模型Hero但response_modelHeroPublic会让 FastAPI 用它来校验与序列化响应数据——secret_name因此被过滤掉。这里特意用response_modelHeroPublic而非返回类型注解- HeroPublic因为实际返回值并非HeroPublic实例。若写- HeroPublic编辑器与 linter 会合理地报错「返回了Hero而非HeroPublic」声明在response_model中则既让 FastAPI 完成序列化又不干扰类型注解对开发工具的帮助。读取配合 HeroPublic 序列化app.get(/heroes/, response_modellist[HeroPublic]) def read_heroes( session: SessionDep, offset: int 0, limit: Annotated[int, Query(le100)] 100, ): heroes session.exec(select(Hero).offset(offset).limit(limit)).all() return heroes app.get(/heroes/{hero_id}, response_modelHeroPublic) def read_hero(hero_id: int, session: SessionDep): hero session.get(Hero, hero_id) if not hero: raise HTTPException(status_code404, detailHero not found) return hero列表用response_modellist[HeroPublic]同样保证逐条校验与序列化。更新PATCH exclude_unsetTrue sqlmodel_updateapp.patch(/heroes/{hero_id}, response_modelHeroPublic) def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep): hero_db session.get(Hero, hero_id) if not hero_db: raise HTTPException(status_code404, detailHero not found) hero_data hero.model_dump(exclude_unsetTrue) hero_db.sqlmodel_update(hero_data) session.add(hero_db) session.commit() session.refresh(hero_db) return hero_db更新的核心技巧有三步通过model_dump(exclude_unsetTrue)拿到仅客户端实际发送的字段字典——被默认值填充、但客户端没提交的字段会被排除这是部分更新的关键 用hero_db.sqlmodel_update(hero_data)把这些变更应用到已存在的hero_db重新add/commit/refresh后返回。对应测试test_tutorial002.py验证了只 PATCH{name: Dog Pond, age: None}时secret_name保持不变、age被显式置空——这正是exclude_unsetTrue区分「未提交」与「显式提交 null」的证据。删除基本不变删除逻辑与单模型版本几乎一致暂不重构app.delete(/heroes/{hero_id}) def delete_hero(hero_id: int, session: SessionDep): hero session.get(Hero, hero_id) if not hero: raise HTTPException(status_code404, detailHero not found) session.delete(hero) session.commit() return {ok: True}运行第二阶段应用$ uv run fastapi dev INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRLC to quit)回到/docs可以看到 API 契约已更新创建英雄时不再要求客户端传id响应中也看不到secret_name。测试佐证从仓库测试看多模型的安全边界仓库的自动化测试精确刻画了多模型版本的预期行为可直接作为理解「为什么安全」的证据test_tutorial002.py客户端 POST 时即使带上id: 9000返回的id也不等于 9000测试断言The ID should be generated by the database——证明客户端无法控制主键响应 JSON 只含name、age、id不含secret_name对 OpenAPI schema 的快照断言显示HeroCreate不要求id、HeroPublic不包含secret_nameHeroUpdate的所有字段均可空test_tutorial002.py。相比之下单模型版本测试test_tutorial001.py的 OpenAPI 快照中请求与响应 schema 都直接引用同一个Hero含id与secret_name——两个版本之间的差异恰好量化了这次「多模型重构」带来的 API 边界收紧效果。测试还揭示了两个实现层面的细节两版测试都把模块的sqlite_url替换为sqlite://内存库并用StaticPool让所有连接共享同一内存数据库同时在重载模块前通过SQLModel.metadata.clear()与default_registry.dispose()清理表模型注册避免跨模块测试污染。小结通过本教程你掌握了用SQLModel与 SQL 数据库交互、并用「数据模型 表模型」精简代码的完整套路表模型tableTrue负责描述数据库表结构主键、索引、列类型数据模型继承但无tableTrue负责描述 API 边界一个全局 engine 每请求一个 Session 依赖既复用了连接又避免了跨请求状态污染角色化模型分层HeroBase→Hero/HeroPublic/HeroCreate/HeroUpdate通过继承复用字段同时把「谁可以写什么、谁能读到什么」固化成 OpenAPI 契约结合 FastAPI 的类型注解与response_model把 SQL 模型直接融入自动文档、请求校验和响应序列化全链路几乎无需手工样板代码。官方教程还提示SQLModel 文档中有一份更完整的「SQLModel 与 FastAPI 配合使用」长篇迷你教程可供继续深入。若要在 FastAPI 项目中使用 SQLModel请以本仓库中 docs_src/sql_databases/tutorial002_an_py310.py 为基线后续再按生产需求加入数据库迁移如 Alembic与独立的数据库服务器。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考