FastAPI 如何用 SQLModel 连接 SQLite 数据库实现数据增删改查？

FastAPI 如何用 SQLModel 连接 SQLite 数据库实现数据增删改查【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi这篇文章解决一个具体的集成任务在 FastAPI 应用中接入 SQL 数据库用 SQLite 作为存储并通过 SQLModel 实现数据的创建、读取、更新和删除。SQLModel 构建在 SQLAlchemy 和 Pydantic 之上是 FastAPI 作者为搭配 FastAPI 开发的库由于它基于 SQLAlchemy凡是 SQLAlchemy 支持的数据库PostgreSQL、MySQL、SQLite、Oracle、Microsoft SQL Server 等都可以使用。文档选 SQLite 做示例是因为它只用单个文件且 Python 有集成支持示例可以原样复制运行生产环境则建议改用 PostgreSQL 这类数据库服务器。安装依赖在项目中加入sqlmodel依赖文档使用 uv 管理项目$ uv add sqlmodel安装完成后应用代码中即可从sqlmodel导入Field、Session、SQLModel、create_engine、select等组件。定义模型、引擎与会话下面以文档中多模型版本的完整代码为基准tutorial002_an_py310.py按顺序说明每个部分的作用。模型table model 与 data modelclass HeroBase(SQLModel): name: str Field(indexTrue) age: int | None Field(defaultNone, indexTrue) class Hero(HeroBase, tableTrue): id: int | None Field(defaultNone, primary_keyTrue) secret_name: str class HeroPublic(HeroBase): id: int class HeroCreate(HeroBase): secret_name: str class HeroUpdate(HeroBase): name: str | None None age: int | None None secret_name: str | None NoneSQLModel 中任何带tableTrue的类都是表模型table model对应数据库中的一张表不带tableTrue的类是数据模型data model本质上就是 Pydantic 模型。文档用继承避免在所有模型中重复声明字段HeroBase存放各模型共享的name、age字段。Hero实际的表模型额外包含id和secret_name。Field(primary_keyTrue)声明主键int | None的写法表示 Python 代码中可以在没有id的情况下创建对象由数据库在保存时生成id而 SQLModel 会在数据库 schema 中把它定义为非空INTEGER列。Field(indexTrue)表示为该列创建 SQL 索引方便按该列过滤读取时加速查找。HeroPublic返回给客户端的模型字段与HeroBase相同因此不包含secret_nameid重新声明为int而非None与 API 客户端形成契约——id一定存在且为int。HeroCreate校验客户端创建数据用的模型包含secret_name。文档指出这正是处理密码的方式接收但不返回且应在使用前哈希、不以明文存储。HeroUpdate更新模型所有字段都是可选的类型包含None且默认值为None这样更新时只需发送要修改的字段。引擎连接 SQLite 的database.dbsqlite_file_name database.db sqlite_url fsqlite:///{sqlite_file_name} connect_args {check_same_thread: False} engine create_engine(sqlite_url, connect_argsconnect_args)SQLModel 的engine底层就是 SQLAlchemyengine负责持有与数据库的连接全部代码应共享同一个engine连接同一个数据库。check_same_threadFalse允许 FastAPI 在不同线程中使用同一个 SQLite 数据库这是必要的因为单个请求可能使用多个线程例如在依赖中。建表函数与每请求一个 Session 的依赖def create_db_and_tables(): SQLModel.metadata.create_all(engine) def get_session(): with Session(engine) as session: yield session SessionDep Annotated[Session, Depends(get_session)] app FastAPI() app.on_event(startup) def on_startup(): create_db_and_tables()SQLModel.metadata.create_all(engine)会为所有表模型创建表文档选择在应用启动事件startup event中执行。文档同时说明生产环境通常应在启动应用之前运行一个迁移脚本来建表SQLModel 后续会提供包装 Alembic 的迁移工具目前可以直接使用 Alembic。Session负责在内存中保存对象、跟踪数据变更并通过engine与数据库通信。这里用yield依赖为每个请求提供一个新的Session保证单请求单会话再用Annotated定义的SessionDep简化后续路由代码。实现增删改查四个操作创建POST用HeroCreate接收、按HeroPublic返回app.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校验在代码中用它构建表模型Heroid由数据库生成session.add→session.commit→session.refresh之后返回。返回处使用response_modelHeroPublic而不是返回类型注解- HeroPublic因为函数实际返回的是Hero对象用response_model声明可以让 FastAPI 用HeroPublic做验证和序列化同时不与编辑器和 linter 的类型提示冲突。读取列表GET支持offset/limit分页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通过select()读取附带limit和offset实现分页limit用Query(le100)限制最大不超过 100。读取单条GET查不到返回 404app.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更新PATCH只更新客户端实际发送的字段app.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更新的关键在exclude_unsetTrue只取出客户端真正发送的数据的字典排除那些仅仅因为默认值而存在的值再用hero_db.sqlmodel_update(hero_data)更新表模型。删除DELETE同样先确认存在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}删除返回{ok: True}。运行应用并验证在应用所在目录启动开发服务器$ uv run fastapi dev文档给出的示例输出文档示例INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRLC to quit)启动后应用会在启动事件中执行create_all首次运行会在当前目录生成database.db文件。验证方式有两个接口文档打开http://127.0.0.1:8000/docs的 Swagger UI。由于模型既是 Pydantic 模型FastAPI 会用这些模型自动生成 API 文档并据此做数据序列化和验证。重构为多模型之后POST /heroes/的请求体示例中不再包含id由数据库生成响应示例也不再返回secret_name在 UI 中Try it out调用各接口POST /heroes/提交{name: …, age: …, secret_name: …}响应返回带id的对象说明记录已写入 SQLiteGET /heroes/返回列表可用offset、limit参数翻页limit上限 100GET /heroes/{hero_id}用刚返回的id查询单条查询不存在的id时返回 404{detail: Hero not found}这是文档中的判断方式PATCH /heroes/{hero_id}只发送要修改的字段例如只发{age: …}其余字段保持不变DELETE /heroes/{hero_id}成功后返回{ok: true}再次GET /heroes/{hero_id}应返回 404。限制与下一步SQLite 仅适合示例/本地开发文档说明 SQLite 之所以被选为示例库是因为单文件、Python 集成支持好生产应用建议换用 PostgreSQL 等数据库服务器。SQLModel 的engineURL 换库即可无需改动模型和路由结构。建表方式有适用边界在 startup 事件中调用create_all只是示例做法文档明确生产环境应在启动应用前运行迁移脚本目前可直接使用 AlembicSQLModel 后续会提供包装 Alembic 的迁移工具。id 不能由客户端决定文档特别指出第一版单模型应用允许客户端在创建时指定id存在覆盖已有id的风险这正是引入HeroCreate/HeroPublic拆分的原因密码类字段应接收后哈希存储不要明文入库。本教程定位为非常简单的入门示例。关于 SQLModel 本身、SQL 语法和更高级的特性文档指向官方 SQLModel 文档其中有一篇更长的 SQLModel FastAPI 迷你教程可继续深入。完整源码可对照仓库中的 tutorial001_an_py310.py单模型版本和 tutorial002_an_py310.py多模型版本教程正文见 docs/en/docs/tutorial/sql-databases.md。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考