从零到可用:完成一个带用户隔离的 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.pyroutersdependencies.pysecurity.pydb.pymodels.pyschemas.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。

从入门项目走向可上线服务

  1. 增加任务完成状态筛选和分页。
  2. 为重复用户名、错误密码、非法 Task 补测试。
  3. 把最终模型接入 Alembic migration
  4. request ID 和统一错误体合入最终项目。
  5. 将最终应用打成 Docker 镜像并使用 PostgreSQL URL 启动。
  6. 增加管理员角色,并写出普通用户越权测试。

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

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

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

写在最后

你已经从第一个 GET 接口走到一个具备类型边界、数据库、认证、授权、测试、观测思路和容器交付路径的 FastAPI 项目。下一步不应继续堆功能,而应选一个真实小需求,保持相同工程标准独立完成:先写接口契约和失败场景,再实现、测试、迁移和交付。