从零到可用:完成一个带用户隔离的 FastAPI 任务 API
摘要:把路由、SQLModel、JWT、依赖注入和 pytest 收束成一个完整任务 API,重点验证未登录访问、双用户数据隔离、部分更新和删除语义。

图:两个用户共享同一套 API,但每次数据库查询都带所有权条件,因此只能操作自己的任务。
最后一个项目不再引入新语法,而是回答一个更接近真实开发的问题:前面那些独立功能组合在一起后,安全边界是否仍然成立?
服务将支持注册、登录和任务 CRUD。Alice 创建的任务对 Bob 不可见,所有者 ID 由服务端从 Token 推导,客户端不能自行指定。最终不是看代码文件齐不齐,而是用一段双用户场景测试证明越权没有发生。
先看完整 API 对外提供哪些能力
最终接口:
| 方法 | 路径 | 是否认证 | 作用 |
|---|---|---|---|
| GET | /health | 否 | 健康检查 |
| POST | /auth/register | 否 | 注册 |
| POST | /auth/token | 否 | 登录取 Token |
| POST | /tasks | 是 | 创建自己的任务 |
| GET | /tasks | 是 | 查询自己的任务 |
| GET/PATCH/DELETE | /tasks/{id} | 是 | 操作自己的任务 |
fastapi dev examples/ch16_final/main.py
先为本地运行准备数据库和签名密钥
cd 04-fastapi-beginner
source .venv/bin/activate
export TASKS_SECRET_KEY="local-development-secret-at-least-32-characters"
首次本地启动会创建 final_tasks.db。测试使用独立内存数据库,不读写该文件。
阅读顺序:app/main.py → routers → dependencies.py → security.py → db.py → models.py 和 schemas.py。
把认证、数据库和任务所有权串起来
数据模型表达所有权
class Task(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
title: str
completed: bool = False
owner_id: int = Field(foreign_key="users.id", index=True)
所有权必须进入数据查询条件,不能只依赖前端隐藏按钮。
当前用户依赖
def get_current_user(token, session):
user_id = decode_access_token(token)
user = session.get(User, user_id)
if user is None:
raise credentials_error
return user
认证成功后,路由获得数据库中的 User 对象,而不是盲目信任 Token 中的任意字段。
强制所有者过滤
statement = select(Task).where(
Task.id == task_id,
Task.owner_id == current_user.id,
)
读取、更新、删除全部复用相同条件。其他用户访问返回 404,避免泄漏任务是否存在。
测试完整旅程
测试会注册 Alice 和 Bob,Alice 创建任务,Bob 列表为空且无法读取 Alice 任务,Alice 可以更新并删除。它验证的是跨接口业务不变量,而不仅是函数覆盖率。

图:任务查询必须同时带上当前用户和所有者条件,不能先查资源再做表面判断。
用真实 Token 走完创建、更新和删除
注册并登录:
curl -X POST http://127.0.0.1:8000/auth/register \
-H "Content-Type: application/json" \
-d '{"username":"alice","password":"safe-password"}'
TOKEN=$(curl -s -X POST http://127.0.0.1:8000/auth/token \
-H "Content-Type: application/x-www-form-urlencoded" \
-d 'username=alice&password=safe-password' \
| python -c "import json,sys; print(json.load(sys.stdin)['access_token'])")
创建、更新和删除:
curl -X POST http://127.0.0.1:8000/tasks \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{"title":"完成 FastAPI 教程"}'
curl -H "Authorization: Bearer $TOKEN" http://127.0.0.1:8000/tasks
curl -X PATCH http://127.0.0.1:8000/tasks/1 \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" -d '{"completed":true}'
curl -i -X DELETE -H "Authorization: Bearer $TOKEN" \
http://127.0.0.1:8000/tasks/1
预期依次为 201、200、200、204。自动重放双用户隔离场景:
python -m pytest tests/test_ch16.py -q

图:双用户旅程既验证交叉访问被拒绝,也验证每个用户能完整操作自己的任务。
安全边界落在每一条数据库查询上
最终依赖链为:Bearer Token → 解码 user_id → Session 查询 User → 路由查询 owner_id 匹配的 Task。任何一环失败,请求都不会进入不安全的数据操作。
数据库模型用于持久化,schema 用于 API 边界;password_hash 只存在内部 User 表模型,不出现在 UserRead。TaskRead 也不要求客户端提交 owner_id,所有者由当前用户依赖决定。
本地示例用 create_all 保持首次运行简单。正式上线时应参考 Alembic 迁移实践替代自动建表,并使用 PostgreSQL、外部密钥和 Docker 交付方案。
用户隔离最容易出现的越权漏洞
- 客户端提交 owner_id:攻击者可以伪造归属;应由服务端从当前用户填写。
- 列表按 owner 过滤,详情忘记过滤:形成越权读取。
- 返回 403 暴露其他用户任务存在:部分系统选择统一 404 降低枚举信息。
- 测试复用开发数据库:数据污染且可能误删真实内容。
- 生产继续使用示例密钥和 create_all:必须迁移到秘密管理和 Alembic。
从入门项目走向可上线服务
- 增加任务完成状态筛选和分页。
- 为重复用户名、错误密码、非法 Task 补测试。
- 把最终模型接入 Alembic migration。
- 将 request ID 和统一错误体合入最终项目。
- 将最终应用打成 Docker 镜像并使用 PostgreSQL URL 启动。
- 增加管理员角色,并写出普通用户越权测试。
最后,用测试固定这次改动
手动请求通过后,运行与本文对应的自动化测试:
python -m pytest tests/test_ch16.py -q
python -m pytest -q
python tests/validate_course.py
写在最后
你已经从第一个 GET 接口走到一个具备类型边界、数据库、认证、授权、测试、观测思路和容器交付路径的 FastAPI 项目。下一步不应继续堆功能,而应选一个真实小需求,保持相同工程标准独立完成:先写接口契约和失败场景,再实现、测试、迁移和交付。