把 FastAPI 内存数据换成 SQLite：SQLModel CRUD 实战

摘要：保持现有任务接口不变，把内存字典替换为 SQLModel 与 SQLite，梳理表模型、请求级 Session、事务提交和测试数据库隔离。

任务数据通过会话和事务写入持久化数据库

图：临时内存数据进入数据库，经请求级会话提交后，即使应用重启仍然存在。

内存 CRUD 最明显的问题是重启即丢数据，更隐蔽的问题是多个进程各有一份字典。接口已经稳定后，下一步不是重写路由，而是把存储层换成真正的关系型数据库。

这里先选 SQLite，因为它不需要单独安装服务，适合观察最小持久化链路。相同的 SQLModel 写法后续可以迁移到 PostgreSQL，路由契约不需要跟着重做。

先验证数据真的跨重启存在

fastapi dev examples/ch09_database/main.py
curl -X POST http://127.0.0.1:8000/tasks \
  -H "Content-Type: application/json" -d '{"title":"写入数据库"}'

停止并重新启动服务后再次查询 /tasks，预期任务仍然存在。

先确认 SQLModel 已经进入虚拟环境

cd 04-fastapi-beginner
source .venv/bin/activate
python -c "import sqlmodel; print(sqlmodel.__version__)"

首次启动会在当前目录生成 tasks.db。它已被 .gitignore 忽略，不会污染版本库。

保持路由不变，只替换存储实现

定义表模型

class Task(TaskBase, table=True):
    __tablename__ = "tasks"
    id: int | None = Field(default=None, primary_key=True)
    completed: bool = Field(default=False, index=True)

table=True 表示类映射到数据库表。创建对象时 ID 暂时为 None，写入数据库后由数据库生成。

提供请求级 Session

def get_session() -> Generator[Session]:
    with Session(engine) as session:
        yield session


SessionDep = Annotated[Session, Depends(get_session)]

yield 前创建资源，响应完成后离开 with 自动关闭。路由无需自行管理连接生命周期。

创建并刷新对象

task = Task.model_validate(payload)
session.add(task)
session.commit()
session.refresh(task)
return task

commit 将事务提交，refresh 从数据库重新加载生成的 ID 等字段。

查询与更新

statement = select(Task).offset(offset).limit(limit)
tasks = session.exec(statement).all()

task.sqlmodel_update(payload.model_dump(exclude_unset=True))
session.add(task)
session.commit()
session.refresh(task)

接口契约不变，通过请求会话切换到数据库存储

图：路由契约保持不变，数据操作通过请求会话进入持久化数据库。

从 API 和 SQLite 两边查看同一条数据

启动并创建两条任务后：

curl http://127.0.0.1:8000/tasks
python - <<'PY'
import sqlite3
connection = sqlite3.connect("tasks.db")
print(connection.execute("select id, title, completed from tasks").fetchall())
connection.close()
PY

预期 Python 命令能直接读到 API 写入的数据。运行隔离测试：

python -m pytest tests/test_ch09.py -q

测试使用 sqlite:// 和 StaticPool，不读写开发数据库文件。

Session 把一次请求组织成工作单元

Session 是一组数据库操作的工作单元。查询得到的对象受 Session 管理，修改后 commit 才形成持久事务。每个请求独立 Session 可以减少请求之间的状态泄漏。

测试使用依赖覆盖把生产 get_session 替换为测试 Session，这正是 Depends 那篇文章里依赖注入的工程价值。接口函数完全不需要知道数据库来自文件还是内存。

create_all() 适合首次学习，但不能安全表达生产表结构演进。增加、重命名、删除字段需要可审查的迁移脚本，下一篇使用 Alembic。

每次请求独立打开、提交并关闭数据库会话

图：一次请求拥有一个独立会话，事务完成后提交，并在响应离开时关闭。

commit、refresh 和测试隔离容易混淆

  • 创建后 ID 仍为 None：忘记 commit 或 refresh。
  • SQLite 跨线程错误：开发示例需要 check_same_thread=False。
  • 测试写入真实 tasks.db：忘记覆盖 get_session。
  • 每个路由手动创建 Session：连接关闭和测试替换容易遗漏。
  • 修改模型后期待旧表自动变化：create_all 不会迁移已有表。

把筛选条件下推到 SQL

  1. 增加 created_at 字段并为创建接口写断言。
  2. 为列表增加 completed 过滤并下推到 SQL where 条件。
  3. 制造一次 commit 前异常，验证数据没有部分写入。
  4. 删除开发数据库重新启动，观察建表流程。
  5. 为数据库 404 增加独立测试。

最后，用测试固定这次改动

手动请求通过后，运行与本文对应的自动化测试：

python -m pytest tests/test_ch09.py -q
python tests/validate_course.py

写在最后

CRUD 已经获得持久化能力，但表结构仍由启动时的 create_all 隐式创建。下一篇将数据库变化变成可版本化、可升级、可回滚的迁移，并展示如何切换 PostgreSQL URL。