欢迎光临
我们一直在努力

从 0 到 1 搭建 API:FastAPI 实战教程(含数据库、部署与安全)

目录

  • 1. 引言
  • 2. 什么是 API
    • 2.1 生活中的类比
    • 2.2 API 的核心组成
  • 3. 搭建前的准备工作
    • 3.1 技术选型
    • 3.2 环境准备
    • 3.3 需求分析
  • 4. 第一个 API 接口
    • 4.1 创建主文件
    • 4.2 启动服务
    • 4.3 测试接口
  • 5. 设计数据模型
    • 5.1 定义模型
    • 5.2 模型说明
  • 6. 实现增删改查
    • 6.1 内存数据存储
    • 6.2 创建待办事项
    • 6.3 查询待办事项列表
    • 6.4 查询单个待办事项
    • 6.5 更新待办事项
    • 6.6 删除待办事项
    • 6.7 全局异常处理
    • 6.8 数据库集成
  • 7. 接口测试与调试
    • 7.1 使用 Swagger 文档测试
    • 7.2 使用 curl 命令测试
    • 7.3 使用 Python requests 调用
    • 7.4 使用 JavaScript fetch 调用
    • 7.5 使用 Postman / Apifox 图形化工具
    • 7.6 常见问题排查
    • 7.7 接口性能对比
    • 7.8 实战测试:完整流程演练
  • 8. 部署上线
    • 8.1 使用 Docker 部署
    • 8.2 部署到云平台
    • 8.3 部署后的监控
  • 9. API 安全
    • 9.1 身份认证
    • 9.2 授权(RBAC)
    • 9.3 输入校验与安全防护
    • 9.4 限流与防暴力破解
    • 9.5 安全最佳实践清单
    • 9.6 HTTPS 与 CORS 安全配置
  • 10. 总结与进阶方向

1. 引言

在当今的软件开发中,API(Application Programming Interface,应用程序编程接口)已经成为连接不同系统、服务和数据的关键桥梁。无论是构建 Web 应用、移动端 App,还是实现微服务架构,API 都扮演着不可或缺的角色。

然而,对于许多初学者来说,API 似乎是一个既熟悉又陌生的概念——每天都在使用别人提供的 API,却不知道如何从零开始设计和搭建一个属于自己的 API。本文将从最基础的概念出发,一步步带你完成一个 API 从设计、开发、测试到部署上线的完整流程。

2. 什么是 API

API 本质上是一组定义了软件组件之间如何交互的规则和协议。它允许不同的软件系统之间进行通信和数据交换,而无需了解对方内部的具体实现细节。

2.1 生活中的类比

可以把 API 想象成餐厅里的服务员:

  • 你(客户端)向服务员(API)点菜(发送请求)
  • 服务员将你的需求传达给厨房(服务器)
  • 厨房做好菜后,服务员再将菜品(响应数据)端到你面前

在这个过程中,你不需要知道厨房内部如何运作,只需要按照菜单(API 文档)点菜即可。

2.2 API 的核心组成

一个典型的 Web API 通常包含以下要素:

  • 端点(Endpoint):API 的访问地址,如 https://api.example.com/users
  • 方法(Method):HTTP 动词,如 GET、POST、PUT、DELETE
  • 参数(Parameters):请求中携带的数据,可以是路径参数、查询参数或请求体
  • 响应(Response):服务器返回的数据,通常为 JSON 或 XML 格式
  • 状态码(Status Code):表示请求结果的数字代码,如 200 表示成功、404 表示资源不存在

3. 搭建前的准备工作

在开始编写代码之前,我们需要先做好环境准备和需求分析。

3.1 技术选型

本文以 Python 的 FastAPI 框架为例进行演示,因为它具有以下优势:

  • 性能优异,基于 Starlette 和 Pydantic
  • 自动生成交互式 API 文档
  • 类型提示友好,代码可读性强
  • 支持异步编程

当然,你也可以选择其他技术栈,如 Node.js 的 Express、Java 的 Spring Boot 等,核心思路是相通的。

3.2 环境准备

首先,确保你的电脑上安装了 Python 3.8 及以上版本,然后创建虚拟环境并安装依赖:

# 创建项目目录
mkdir my-api
cd my-api

# 创建虚拟环境
python -m venv venv

# 激活虚拟环境(Windows)
venv\\Scripts\\activate

# 激活虚拟环境(macOS/Linux)
source venv/bin/activate

# 安装 FastAPI 和 Uvicorn
pip install fastapi uvicorn

3.3 需求分析

在动手写代码之前,先明确我们的 API 要解决什么问题。本文以一个简单的「待办事项管理」API 为例,需要实现以下功能:

  • 创建待办事项
  • 查询待办事项列表
  • 查询单个待办事项详情
  • 更新待办事项
  • 删除待办事项

4. 第一个 API 接口

现在,让我们从最简单的接口开始,感受一下 FastAPI 的开发体验。

4.1 创建主文件

在项目目录下创建 main.py 文件:

from fastapi import FastAPI

# 创建 FastAPI 实例
app = FastAPI(title="待办事项 API", version="1.0.0")

@app.get("/")
def read_root():
return {"message": "欢迎使用待办事项 API"}

4.2 启动服务

在终端中运行以下命令启动开发服务器:

uvicorn main:app –reload

启动成功后,你会看到类似如下的输出:

INFO: Uvicorn running on http://127.0.0.1:8000
INFO: Application startup complete.

4.3 测试接口

打开浏览器访问 http://127.0.0.1:8000,你会看到返回的 JSON 数据:

{
"message": "欢迎使用待办事项 API"
}

同时,FastAPI 还自动生成了交互式 API 文档,访问 http://127.0.0.1:8000/docs 即可查看和在线调试所有接口。

5. 设计数据模型

接下来,我们使用 Pydantic 定义待办事项的数据模型。Pydantic 是 FastAPI 的数据验证核心,可以自动完成请求数据的类型校验和序列化。

5.1 定义模型

在 main.py 中新增数据模型:

from typing import Optional
from pydantic import BaseModel
from datetime import datetime

class TodoItem(BaseModel):
"""待办事项数据模型"""
id: Optional[int] = None
title: str
description: Optional[str] = None
completed: bool = False
created_at: datetime = datetime.now()

5.2 模型说明

  • id:待办事项的唯一标识,创建时自动生成
  • title:标题,必填字段
  • description:详细描述,可选字段
  • completed:是否已完成,默认为 False
  • created_at:创建时间,默认为当前时间

6. 实现增删改查

现在,我们来实现完整的 CRUD(Create、Read、Update、Delete)接口。

6.1 内存数据存储

为了简化演示,我们先使用内存列表存储数据。在实际项目中,你会使用数据库(如 PostgreSQL、MySQL)来持久化数据。

# 内存数据存储
todos = []
todo_id_counter = 1

6.2 创建待办事项

from fastapi import HTTPException

@app.post("/todos", status_code=201)
def create_todo(todo: TodoItem):
"""创建新的待办事项"""
global todo_id_counter
todo.id = todo_id_counter
todo_id_counter += 1
todos.append(todo)
return todo

6.3 查询待办事项列表

@app.get("/todos")
def list_todos(completed: Optional[bool] = None):
"""获取待办事项列表,可按完成状态筛选"""
if completed is None:
return todos
return [todo for todo in todos if todo.completed == completed]

6.4 查询单个待办事项

@app.get("/todos/{todo_id}")
def get_todo(todo_id: int):
"""根据 ID 获取单个待办事项"""
for todo in todos:
if todo.id == todo_id:
return todo
raise HTTPException(status_code=404, detail="待办事项不存在")

6.5 更新待办事项

@app.put("/todos/{todo_id}")
def update_todo(todo_id: int, todo_update: TodoItem):
"""更新指定 ID 的待办事项"""
for index, todo in enumerate(todos):
if todo.id == todo_id:
# 保留原 ID 和创建时间
todo_update.id = todo.id
todo_update.created_at = todo.created_at
todos[index] = todo_update
return todo_update
raise HTTPException(status_code=404, detail="待办事项不存在")

6.6 删除待办事项

@app.delete("/todos/{todo_id}", status_code=204)
def delete_todo(todo_id: int):
"""删除指定 ID 的待办事项"""
for index, todo in enumerate(todos):
if todo.id == todo_id:
todos.pop(index)
return
raise HTTPException(status_code=404, detail="待办事项不存在")

删除成功后返回 204 No Content,表示操作成功且无返回内容。如果待办事项不存在,则返回 404 错误。

6.7 全局异常处理

6.8 数据库集成

前面我们使用内存列表存储数据,服务重启后数据就会丢失。下面我们使用 SQLAlchemy 连接 SQLite 数据库,实现数据的持久化存储。

6.8.1 安装依赖

pip install sqlalchemy

6.8.2 配置数据库

在 main.py 中新增数据库配置和模型:

from sqlalchemy import create_engine, Column, Integer, String, Boolean, DateTime
from sqlalchemy.orm import declarative_base, sessionmaker
from datetime import datetime

# 数据库配置(使用 SQLite,实际项目可替换为 PostgreSQL/MySQL)
DATABASE_URL = "sqlite:///./todos.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class TodoDB(Base):
"""数据库中的待办事项表"""
__tablename__ = "todos"

id = Column(Integer, primary_key=True, index=True)
title = Column(String, nullable=False)
description = Column(String, nullable=True)
completed = Column(Boolean, default=False)
created_at = Column(DateTime, default=datetime.now)

# 创建数据表
Base.metadata.create_all(bind=engine)

6.8.3 改造 CRUD 接口

将原来的内存操作替换为数据库操作:

from fastapi import Depends
from sqlalchemy.orm import Session

def get_db():
"""获取数据库会话"""
db = SessionLocal()
try:
yield db
finally:
db.close()

@app.post("/todos", status_code=201)
def create_todo(todo: TodoItem, db: Session = Depends(get_db)):
"""创建新的待办事项(写入数据库)"""
db_todo = TodoDB(
title=todo.title,
description=todo.description,
completed=todo.completed,
)
db.add(db_todo)
db.commit()
db.refresh(db_todo)
return db_todo

@app.get("/todos")
def list_todos(completed: Optional[bool] = None, db: Session = Depends(get_db)):
"""获取待办事项列表(从数据库读取)"""
query = db.query(TodoDB)
if completed is not None:
query = query.filter(TodoDB.completed == completed)
return query.all()

@app.get("/todos/{todo_id}")
def get_todo(todo_id: int, db: Session = Depends(get_db)):
"""根据 ID 获取单个待办事项"""
todo = db.query(TodoDB).filter(TodoDB.id == todo_id).first()
if not todo:
raise HTTPException(status_code=404, detail="待办事项不存在")
return todo

@app.put("/todos/{todo_id}")
def update_todo(todo_id: int, todo_update: TodoItem, db: Session = Depends(get_db)):
"""更新指定 ID 的待办事项"""
todo = db.query(TodoDB).filter(TodoDB.id == todo_id).first()
if not todo:
raise HTTPException(status_code=404, detail="待办事项不存在")
todo.title = todo_update.title
todo.description = todo_update.description
todo.completed = todo_update.completed
db.commit()
db.refresh(todo)
return todo

@app.delete("/todos/{todo_id}", status_code=204)
def delete_todo(todo_id: int, db: Session = Depends(get_db)):
"""删除指定 ID 的待办事项"""
todo = db.query(TodoDB).filter(TodoDB.id == todo_id).first()
if not todo:
raise HTTPException(status_code=404, detail="待办事项不存在")
db.delete(todo)
db.commit()

改造完成后,重启服务,数据就会持久化保存到 todos.db 文件中,即使服务重启数据也不会丢失。

为了让 API 在出错时返回统一、友好的错误信息,我们添加一个全局异常处理器。这样无论是参数校验失败、资源不存在,还是服务器内部错误,客户端都能收到结构一致的 JSON 响应。

from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
"""统一处理 HTTP 异常,返回 JSON 格式的错误信息"""
return JSONResponse(
status_code=exc.status_code,
content={"code": exc.status_code, "message": exc.detail, "data": None},
)

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
"""统一处理请求参数校验失败,返回字段级别的错误详情"""
errors = exc.errors()
detail = []
for error in errors:
field = ".".join(str(loc) for loc in error["loc"] if loc != "body")
detail.append({"field": field, "message": error["msg"]})
return JSONResponse(
status_code=422,
content={"code": 422, "message": "请求参数校验失败", "data": detail},
)

@app.exception_handler(Exception)
async def unhandled_exception_handler(request: Request, exc: Exception):
"""兜底捕获未处理的异常,避免向客户端泄露堆栈信息"""
return JSONResponse(
status_code=500,
content={"code": 500, "message": "服务器内部错误,请稍后重试", "data": None},
)

添加后,所有接口的错误响应都会统一为 {"code": …, "message": …, "data": …} 的结构,前端解析起来更加方便。

@app.delete("/todos/{todo_id}", status_code=204)
def delete_todo(todo_id: int):
"""删除指定 ID 的待办事项"""
for index, todo in enumerate(todos):
if todo.id == todo_id:
todos.pop(index)
return
raise HTTPException(status_code=404, detail="待办事项不存在")

7. 接口测试与调试

接口写完后,我们需要进行充分的测试,确保每个接口都能正常工作。除了 Swagger 和 curl,API 还支持多种调用方式,下面逐一介绍。

7.1 使用 Swagger 文档测试

FastAPI 自动生成的 /docs 页面提供了可视化的接口测试工具。你可以:

  • 打开 http://127.0.0.1:8000/docs
  • 点击任意接口展开详情
  • 点击「Try it out」按钮
  • 填写参数后点击「Execute」发送请求
  • 查看响应结果和状态码
  • 7.2 使用 curl 命令测试

    在终端中也可以直接使用 curl 进行测试:

    # 创建待办事项
    curl -X POST "http://127.0.0.1:8000/todos" \\
    -H "Content-Type: application/json" \\
    -d '{"title": "学习 FastAPI", "description": "完成 API 开发教程"}'

    # 查询列表
    curl "http://127.0.0.1:8000/todos"

    # 查询单个
    curl "http://127.0.0.1:8000/todos/1"

    # 更新
    curl -X PUT "http://127.0.0.1:8000/todos/1" \\
    -H "Content-Type: application/json" \\
    -d '{"title": "学习 FastAPI", "description": "已完成", "completed": true}'

    # 删除
    curl -X DELETE "http://127.0.0.1:8000/todos/1"

    7.3 使用 Python requests 调用

    在 Python 脚本或 Jupyter Notebook 中,可以使用 requests 库调用 API:

    import requests

    BASE_URL = "http://127.0.0.1:8000"

    # 创建待办事项
    resp = requests.post(
    f"{BASE_URL}/todos",
    json={"title": "学习 FastAPI", "description": "完成 API 开发教程"},
    )
    print(resp.status_code, resp.json())

    # 查询列表
    resp = requests.get(f"{BASE_URL}/todos")
    print(resp.status_code, resp.json())

    # 查询单个
    resp = requests.get(f"{BASE_URL}/todos/1")
    print(resp.status_code, resp.json())

    # 更新
    resp = requests.put(
    f"{BASE_URL}/todos/1",
    json={"title": "学习 FastAPI", "description": "已完成", "completed": True},
    )
    print(resp.status_code, resp.json())

    # 删除
    resp = requests.delete(f"{BASE_URL}/todos/1")
    print(resp.status_code)

    7.4 使用 JavaScript fetch 调用

    在前端浏览器或 Node.js 环境中,可以使用 fetch 调用 API:

    const BASE_URL = "http://127.0.0.1:8000";

    // 创建待办事项
    const createResp = await fetch(`${BASE_URL}/todos`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
    title: "学习 FastAPI",
    description: "完成 API 开发教程",
    }),
    });
    console.log(createResp.status, await createResp.json());

    // 查询列表
    const listResp = await fetch(`${BASE_URL}/todos`);
    console.log(listResp.status, await listResp.json());

    // 查询单个
    const getResp = await fetch(`${BASE_URL}/todos/1`);
    console.log(getResp.status, await getResp.json());

    // 更新
    const updateResp = await fetch(`${BASE_URL}/todos/1`, {
    method: "PUT",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
    title: "学习 FastAPI",
    description: "已完成",
    completed: true,
    }),
    });
    console.log(updateResp.status, await updateResp.json());

    // 删除
    const deleteResp = await fetch(`${BASE_URL}/todos/1`, { method: "DELETE" });
    console.log(deleteResp.status);

    7.5 使用 Postman / Apifox 图形化工具

    Postman 和 Apifox 是流行的图形化 API 调试工具,适合团队协作和接口文档管理:

  • 新建请求,选择方法(GET/POST/PUT/DELETE)
  • 填写 URL,如 http://127.0.0.1:8000/todos
  • 在 Body 选项卡选择 raw + JSON,填入请求体
  • 点击 Send 发送请求,查看响应结果
  • 可将接口保存到集合中,方便复用和分享
  • 调用方式选择建议:日常调试用 Swagger 或 Postman;脚本自动化用 Python requests;前端联调用 fetch 或 axios;命令行快速验证用 curl。

    接口写完后,我们需要进行充分的测试,确保每个接口都能正常工作。

    7.1 使用 Swagger 文档测试

    FastAPI 自动生成的 /docs 页面提供了可视化的接口测试工具。你可以:

  • 打开 http://127.0.0.1:8000/docs
  • 点击任意接口展开详情
  • 点击「Try it out」按钮
  • 填写参数后点击「Execute」发送请求
  • 查看响应结果和状态码
  • 7.2 使用 curl 命令测试

    在终端中也可以直接使用 curl 进行测试:

    # 创建待办事项
    curl -X POST "http://127.0.0.1:8000/todos" \\
    -H "Content-Type: application/json" \\
    -d '{"title": "学习 FastAPI", "description": "完成 API 开发教程"}'

    # 查询列表
    curl "http://127.0.0.1:8000/todos"

    # 查询单个
    curl "http://127.0.0.1:8000/todos/1"

    # 更新
    curl -X PUT "http://127.0.0.1:8000/todos/1" \\
    -H "Content-Type: application/json" \\
    -d '{"title": "学习 FastAPI", "description": "已完成", "completed": true}'

    # 删除
    curl -X DELETE "http://127.0.0.1:8000/todos/1"

    7.3 常见问题排查

    在开发和测试过程中,你可能会遇到一些常见问题。下面按错误类型给出排查思路和解决方案。

    7.3.1 404 错误:资源不存在

    现象:请求返回 404 Not Found。

    可能原因与解决方案:

    • URL 路径写错:检查路径是否与路由定义完全一致,注意大小写和末尾斜杠
    • 资源 ID 不存在:确认查询的 ID 是否在数据库中存在,可先调用列表接口确认
    • 路由顺序问题:如果定义了 /todos/{todo_id} 和 /todos/me 这类路由,注意静态路径要放在动态路径之前
    7.3.2 422 错误:参数校验失败

    现象:请求返回 422 Unprocessable Entity,响应体中包含字段级别的错误详情。

    可能原因与解决方案:

    • 缺少必填字段:如创建待办事项时未传 title,检查请求体是否完整
    • 字段类型错误:如 completed 传了字符串 "true" 而不是布尔值 true,检查 JSON 中的类型
    • 路径参数类型错误:如 /todos/abc 中 abc 无法转换为 int,确认路径参数类型
    7.3.3 500 错误:服务器内部异常

    现象:请求返回 500 Internal Server Error。

    可能原因与解决方案:

    • 数据库连接失败:检查数据库服务是否启动、连接串是否正确
    • 代码逻辑异常:查看终端日志中的堆栈信息,定位具体报错行
    • 依赖缺失:确认所有依赖都已安装,可执行 pip list 检查
    7.3.4 服务无法启动

    现象:运行 uvicorn main:app –reload 时报错退出。

    可能原因与解决方案:

    • 端口被占用:换一个端口启动,如 uvicorn main:app –port 8001
    • 导入错误:检查 main.py 中是否有语法错误或未安装的依赖
    • 虚拟环境未激活:确认当前终端使用的是项目虚拟环境
    7.3.5 数据不持久化

    现象:服务重启后,之前创建的数据丢失。

    可能原因与解决方案:

    • 仍在使用内存存储:确认是否已按第 6.8 节改造为数据库存储
    • 数据库文件路径问题:检查 DATABASE_URL 中的路径是否正确,SQLite 文件是否生成
    • 未执行建表:确认 Base.metadata.create_all(bind=engine) 已执行

    排查技巧:遇到问题时,先看终端日志中的堆栈信息,再结合响应体中的 code 和 message 字段定位问题,通常能快速找到根因。

    7.4 接口性能对比

    为了直观展示不同实现方式的性能差异,下面用一张对比图说明内存存储与数据库存储、同步与异步接口、有无缓存等场景下的响应耗时情况。

    #mermaid-svg-iJbumWrc2scx5FVY{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-iJbumWrc2scx5FVY .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-iJbumWrc2scx5FVY .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-iJbumWrc2scx5FVY .error-icon{fill:#552222;}#mermaid-svg-iJbumWrc2scx5FVY .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-iJbumWrc2scx5FVY .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-iJbumWrc2scx5FVY .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-iJbumWrc2scx5FVY .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-iJbumWrc2scx5FVY .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-iJbumWrc2scx5FVY .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-iJbumWrc2scx5FVY .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-iJbumWrc2scx5FVY .marker{fill:#333333;stroke:#333333;}#mermaid-svg-iJbumWrc2scx5FVY .marker.cross{stroke:#333333;}#mermaid-svg-iJbumWrc2scx5FVY svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-iJbumWrc2scx5FVY p{margin:0;}#mermaid-svg-iJbumWrc2scx5FVY .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-iJbumWrc2scx5FVY .cluster-label text{fill:#333;}#mermaid-svg-iJbumWrc2scx5FVY .cluster-label span{color:#333;}#mermaid-svg-iJbumWrc2scx5FVY .cluster-label span p{background-color:transparent;}#mermaid-svg-iJbumWrc2scx5FVY .label text,#mermaid-svg-iJbumWrc2scx5FVY span{fill:#333;color:#333;}#mermaid-svg-iJbumWrc2scx5FVY .node rect,#mermaid-svg-iJbumWrc2scx5FVY .node circle,#mermaid-svg-iJbumWrc2scx5FVY .node ellipse,#mermaid-svg-iJbumWrc2scx5FVY .node polygon,#mermaid-svg-iJbumWrc2scx5FVY .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-iJbumWrc2scx5FVY .rough-node .label text,#mermaid-svg-iJbumWrc2scx5FVY .node .label text,#mermaid-svg-iJbumWrc2scx5FVY .image-shape .label,#mermaid-svg-iJbumWrc2scx5FVY .icon-shape .label{text-anchor:middle;}#mermaid-svg-iJbumWrc2scx5FVY .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-iJbumWrc2scx5FVY .rough-node .label,#mermaid-svg-iJbumWrc2scx5FVY .node .label,#mermaid-svg-iJbumWrc2scx5FVY .image-shape .label,#mermaid-svg-iJbumWrc2scx5FVY .icon-shape .label{text-align:center;}#mermaid-svg-iJbumWrc2scx5FVY .node.clickable{cursor:pointer;}#mermaid-svg-iJbumWrc2scx5FVY .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-iJbumWrc2scx5FVY .arrowheadPath{fill:#333333;}#mermaid-svg-iJbumWrc2scx5FVY .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-iJbumWrc2scx5FVY .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-iJbumWrc2scx5FVY .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-iJbumWrc2scx5FVY .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-iJbumWrc2scx5FVY .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-iJbumWrc2scx5FVY .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-iJbumWrc2scx5FVY .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-iJbumWrc2scx5FVY .cluster text{fill:#333;}#mermaid-svg-iJbumWrc2scx5FVY .cluster span{color:#333;}#mermaid-svg-iJbumWrc2scx5FVY div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-iJbumWrc2scx5FVY .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-iJbumWrc2scx5FVY rect.text{fill:none;stroke-width:0;}#mermaid-svg-iJbumWrc2scx5FVY .icon-shape,#mermaid-svg-iJbumWrc2scx5FVY .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-iJbumWrc2scx5FVY .icon-shape p,#mermaid-svg-iJbumWrc2scx5FVY .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-iJbumWrc2scx5FVY .icon-shape .label rect,#mermaid-svg-iJbumWrc2scx5FVY .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-iJbumWrc2scx5FVY .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-iJbumWrc2scx5FVY .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-iJbumWrc2scx5FVY :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

    适用场景

    缓存命中率

    平均响应耗时

    接口实现方式

    内存存储(无 I/O)

    SQLite 存储(本地磁盘 I/O)

    PostgreSQL 存储(网络 I/O)

    SQLite + Redis 缓存(缓存命中)

    异步 PostgreSQL(asyncpg)

    0.1ms – 0.5ms

    1ms – 5ms

    5ms – 20ms

    0.2ms – 1ms

    3ms – 10ms

    100%(无缓存)

    0%(无缓存)

    0%(无缓存)

    90% – 99%

    0%(无缓存)

    缓存 / 临时数据

    中小型应用 / 开发环境

    生产环境 / 高并发

    高频读取 / 低频更新

    高并发写入 / 大数据量

    从图中可以看出:

    • 内存存储:响应最快,但数据无法持久化,适合缓存或临时数据
    • SQLite 存储:本地磁盘 I/O,速度尚可,适合中小型应用和开发环境
    • PostgreSQL 存储:网络 I/O 开销较大,但支持并发、事务和水平扩展,适合生产环境
    • SQLite + Redis 缓存:缓存命中时响应接近内存速度,适合高频读取、低频更新的数据
    • 异步 PostgreSQL:通过异步驱动减少线程阻塞,在高并发下吞吐量显著提升

    读多写少场景:优先考虑加缓存;写多读少场景:优先考虑异步驱动和连接池;高并发场景:两者结合效果最佳。

    7.4.1 性能对比数据表

    为了更直观地对比不同实现方式的性能差异,下面给出一个参考数据表(基于 1000 次请求的平均值):

    实现方式平均响应耗时吞吐量(QPS)缓存命中率数据持久化适用场景
    内存存储 0.1ms – 0.5ms 5000+ 100%(无缓存) 缓存、临时数据
    SQLite 存储 1ms – 5ms 800 – 1500 0%(无缓存) 中小型应用、开发环境
    PostgreSQL 存储 5ms – 20ms 300 – 800 0%(无缓存) 生产环境、高并发
    SQLite + Redis 缓存 0.2ms – 1ms 3000+ 90% – 99% 高频读取、低频更新
    异步 PostgreSQL 3ms – 10ms 1000 – 3000 0%(无缓存) 高并发写入、大数据量

    缓存命中率说明:缓存命中率越高,实际响应越接近内存速度。当缓存命中率低于 80% 时,建议检查缓存淘汰策略或调整过期时间,避免缓存穿透和雪崩。

    7.5 实战测试:完整流程演练

    下面通过一个完整的实战演练,把前面学到的接口串起来跑一遍,验证 CRUD 全流程的正确性。

    7.5.1 准备测试数据

    先创建两条待办事项作为测试数据:

    # 创建第一条待办事项
    curl -X POST "http://127.0.0.1:8000/todos" \\
    -H "Content-Type: application/json" \\
    -d '{"title": "学习 FastAPI", "description": "完成 API 开发教程"}'

    # 创建第二条待办事项
    curl -X POST "http://127.0.0.1:8000/todos" \\
    -H "Content-Type: application/json" \\
    -d '{"title": "编写单元测试", "description": "为 CRUD 接口补充测试用例"}'

    7.5.2 验证查询功能

    # 查询列表,应返回 2 条记录
    curl "http://127.0.0.1:8000/todos"

    # 按完成状态筛选
    curl "http://127.0.0.1:8000/todos?completed=false"

    # 查询单个
    curl "http://127.0.0.1:8000/todos/1"

    7.5.3 验证更新功能

    # 将第一条待办事项标记为已完成
    curl -X PUT "http://127.0.0.1:8000/todos/1" \\
    -H "Content-Type: application/json" \\
    -d '{"title": "学习 FastAPI", "description": "完成 API 开发教程", "completed": true}'

    7.5.4 验证删除功能

    # 删除第二条待办事项
    curl -X DELETE "http://127.0.0.1:8000/todos/2"

    # 再次查询列表,应只剩 1 条记录
    curl "http://127.0.0.1:8000/todos"

    7.5.5 验证错误处理

    # 查询不存在的 ID,应返回 404
    curl "http://127.0.0.1:8000/todos/999"

    # 缺少必填字段,应返回 422
    curl -X POST "http://127.0.0.1:8000/todos" \\
    -H "Content-Type: application/json" \\
    -d '{"description": "缺少 title 字段"}'

    7.5.6 使用 Python 脚本自动化测试

    手动测试只能验证单个接口,下面用 Python 脚本把整个流程自动化跑一遍,并断言每个步骤的返回结果:

    import requests

    BASE_URL = "http://127.0.0.1:8000"

    def test_create_todo():
    """测试创建待办事项"""
    resp = requests.post(
    f"{BASE_URL}/todos",
    json={"title": "学习 FastAPI", "description": "完成 API 开发教程"},
    )
    assert resp.status_code == 201, f"创建失败: {resp.status_code}"
    data = resp.json()
    assert data["title"] == "学习 FastAPI"
    print(f"✅ 创建成功: id={data['id']}")
    return data["id"]

    def test_list_todos():
    """测试查询列表"""
    resp = requests.get(f"{BASE_URL}/todos")
    assert resp.status_code == 200
    assert isinstance(resp.json(), list)
    print(f"✅ 查询列表成功: 共 {len(resp.json())} 条")

    def test_get_todo(todo_id: int):
    """测试查询单个"""
    resp = requests.get(f"{BASE_URL}/todos/{todo_id}")
    assert resp.status_code == 200
    assert resp.json()["id"] == todo_id
    print(f"✅ 查询单个成功: id={todo_id}")

    def test_update_todo(todo_id: int):
    """测试更新"""
    resp = requests.put(
    f"{BASE_URL}/todos/{todo_id}",
    json={"title": "学习 FastAPI", "description": "已完成", "completed": True},
    )
    assert resp.status_code == 200
    assert resp.json()["completed"] is True
    print(f"✅ 更新成功: id={todo_id}")

    def test_delete_todo(todo_id: int):
    """测试删除"""
    resp = requests.delete(f"{BASE_URL}/todos/{todo_id}")
    assert resp.status_code == 204
    print(f"✅ 删除成功: id={todo_id}")

    def test_not_found():
    """测试 404 错误处理"""
    resp = requests.get(f"{BASE_URL}/todos/99999")
    assert resp.status_code == 404
    assert resp.json()["code"] == 404
    print("✅ 404 错误处理正确")

    def test_validation_error():
    """测试 422 参数校验错误"""
    resp = requests.post(f"{BASE_URL}/todos", json={"description": "缺少 title"})
    assert resp.status_code == 422
    assert resp.json()["code"] == 422
    print("✅ 422 参数校验错误处理正确")

    if __name__ == "__main__":
    print("开始自动化测试…\\n")
    todo_id = test_create_todo()
    test_list_todos()
    test_get_todo(todo_id)
    test_update_todo(todo_id)
    test_delete_todo(todo_id)
    test_not_found()
    test_validation_error()
    print("\\n🎉 全部测试通过!")

    运行脚本:

    python test_api.py

    预期输出:

    开始自动化测试…

    ✅ 创建成功: id=1
    ✅ 查询列表成功: 共 1 条
    ✅ 查询单个成功: id=1
    ✅ 更新成功: id=1
    ✅ 删除成功: id=1
    ✅ 404 错误处理正确
    ✅ 422 参数校验错误处理正确

    🎉 全部测试通过!

    测试要点:自动化测试脚本应覆盖正常流程(增删改查)和异常流程(404、422),并断言状态码和关键字段。实际项目中可结合 pytest 编写更完整的测试套件,并接入 CI/CD 流水线。

    7.5.7 使用 pytest 编写自动化测试

    除了手动调用脚本,更推荐使用 pytest 编写可重复执行的自动化测试。这样每次修改代码后,只需运行一条命令即可验证所有接口是否正常。

    首先安装 pytest 和 httpx(FastAPI 的测试客户端依赖):

    pip install pytest httpx

    然后创建 test_main.py 测试文件:

    import pytest
    from fastapi.testclient import TestClient
    from main import app

    client = TestClient(app)

    def test_create_todo():
    """测试创建待办事项"""
    resp = client.post("/todos", json={
    "title": "学习 pytest",
    "description": "编写自动化测试",
    })
    assert resp.status_code == 201
    data = resp.json()
    assert data["title"] == "学习 pytest"
    assert data["completed"] is False
    assert "id" in data

    def test_list_todos():
    """测试查询待办事项列表"""
    # 先创建一条数据
    client.post("/todos", json={"title": "测试列表"})

    resp = client.get("/todos")
    assert resp.status_code == 200
    assert isinstance(resp.json(), list)
    assert len(resp.json()) >= 1

    def test_get_todo_not_found():
    """测试查询不存在的待办事项返回 404"""
    resp = client.get("/todos/99999")
    assert resp.status_code == 404
    assert resp.json()["code"] == 404

    def test_update_todo():
    """测试更新待办事项"""
    # 先创建一条数据
    create_resp = client.post("/todos", json={"title": "待更新"})
    todo_id = create_resp.json()["id"]

    resp = client.put(f"/todos/{todo_id}", json={
    "title": "已更新",
    "completed": True,
    })
    assert resp.status_code == 200
    assert resp.json()["title"] == "已更新"
    assert resp.json()["completed"] is True

    def test_delete_todo():
    """测试删除待办事项"""
    # 先创建一条数据
    create_resp = client.post("/todos", json={"title": "待删除"})
    todo_id = create_resp.json()["id"]

    resp = client.delete(f"/todos/{todo_id}")
    assert resp.status_code == 204

    # 再次查询应返回 404
    resp = client.get(f"/todos/{todo_id}")
    assert resp.status_code == 404

    def test_validation_error():
    """测试参数校验失败返回 422"""
    resp = client.post("/todos", json={"title": ""})
    assert resp.status_code == 422
    assert resp.json()["code"] == 422

    运行测试:

    pytest test_main.py -v

    输出示例:

    ============================= test session starts =============================
    collected 6 items

    test_main.py::test_create_todo PASSED
    test_main.py::test_list_todos PASSED
    test_main.py::test_get_todo_not_found PASSED
    test_main.py::test_update_todo PASSED
    test_main.py::test_delete_todo PASSED
    test_main.py::test_validation_error PASSED

    ============================== 6 passed in 0.35s ==============================

    测试要点:使用 TestClient 无需启动真实服务器即可模拟请求;每个测试用例相互独立,建议在测试前清空数据库或使用独立测试库;断言时同时检查状态码和响应体,确保接口行为符合预期。

    7.5.8 使用 pytest 测试数据库集成

    在完成数据库改造后,我们需要验证 CRUD 接口与 SQLite 的集成是否正常。下面是一个针对数据库版本的 pytest 测试脚本:

    # test_db_integration.py
    import os
    import tempfile
    import pytest
    from fastapi.testclient import TestClient

    # 使用临时数据库,避免污染开发数据
    os.environ["DATABASE_URL"] = f"sqlite:///{tempfile.mkdtemp()}/test.db"

    from main import app, Base, engine, SessionLocal, TodoDB

    # 创建测试表
    Base.metadata.create_all(bind=engine)

    client = TestClient(app)

    @pytest.fixture(autouse=True)
    def clean_db():
    """每个测试用例前清空数据库"""
    db = SessionLocal()
    db.query(TodoDB).delete()
    db.commit()
    db.close()
    yield

    def test_create_todo_persists_to_db():
    """创建待办事项后,数据应写入数据库"""
    resp = client.post("/todos", json={"title": "数据库测试", "description": "验证持久化"})
    assert resp.status_code == 201
    data = resp.json()
    assert data["title"] == "数据库测试"

    # 直接从数据库验证
    db = SessionLocal()
    todo = db.query(TodoDB).filter(TodoDB.id == data["id"]).first()
    assert todo is not None
    assert todo.title == "数据库测试"
    db.close()

    def test_list_todos_reads_from_db():
    """列表接口应返回数据库中的数据"""
    # 先创建一条数据
    client.post("/todos", json={"title": "待查询事项"})

    resp = client.get("/todos")
    assert resp.status_code == 200
    assert len(resp.json()) == 1
    assert resp.json()[0]["title"] == "待查询事项"

    def test_update_todo_syncs_to_db():
    """更新接口应同步修改数据库记录"""
    create_resp = client.post("/todos", json={"title": "原始标题"})
    todo_id = create_resp.json()["id"]

    update_resp = client.put(
    f"/todos/{todo_id}",
    json={"title": "更新后的标题", "completed": True},
    )
    assert update_resp.status_code == 200
    assert update_resp.json()["title"] == "更新后的标题"

    # 验证数据库中的记录已更新
    db = SessionLocal()
    todo = db.query(TodoDB).filter(TodoDB.id == todo_id).first()
    assert todo.title == "更新后的标题"
    assert todo.completed is True
    db.close()

    def test_delete_todo_removes_from_db():
    """删除接口应移除数据库记录"""
    create_resp = client.post("/todos", json={"title": "待删除事项"})
    todo_id = create_resp.json()["id"]

    delete_resp = client.delete(f"/todos/{todo_id}")
    assert delete_resp.status_code == 204

    # 验证数据库中的记录已删除
    db = SessionLocal()
    todo = db.query(TodoDB).filter(TodoDB.id == todo_id).first()
    assert todo is None
    db.close()

    运行测试:

    pytest test_db_integration.py -v

    测试要点:使用 TestClient 模拟真实 HTTP 请求,无需启动服务器;通过 autouse fixture 在每个用例前清空数据库,保证测试隔离;测试结束后临时数据库自动清理,不影响开发数据。

    7.5.9 使用 pytest 测试部署后的接口

    部署完成后,我们同样可以使用 pytest 对线上接口进行冒烟测试,确保服务在云平台上正常运行。下面是一个针对部署环境的测试脚本:

    # test_deployed_api.py
    """
    部署环境冒烟测试
    验证线上 API 的核心功能是否正常
    """

    import os
    import requests
    import pytest

    # 从环境变量读取线上地址,本地测试时使用默认值
    BASE_URL = os.getenv("DEPLOYED_API_URL", "https://your-app.railway.app")

    @pytest.fixture(scope="module")
    def base_url():
    """返回线上 API 地址"""
    return BASE_URL

    def test_health_check(base_url):
    """健康检查接口应返回 200"""
    resp = requests.get(f"{base_url}/health", timeout=10)
    assert resp.status_code == 200
    data = resp.json()
    assert data["status"] == "ok"

    def test_create_todo(base_url):
    """线上创建待办事项"""
    resp = requests.post(
    f"{base_url}/todos",
    json={"title": "部署冒烟测试", "description": "验证线上创建功能"},
    timeout=10,
    )
    assert resp.status_code == 201
    data = resp.json()
    assert data["title"] == "部署冒烟测试"
    return data["id"]

    def test_list_todos(base_url):
    """线上查询列表"""
    resp = requests.get(f"{base_url}/todos", timeout=10)
    assert resp.status_code == 200
    assert isinstance(resp.json(), list)

    def test_get_todo(base_url):
    """线上查询单个待办事项"""
    # 先创建一个
    create_resp = requests.post(
    f"{base_url}/todos",
    json={"title": "查询测试"},
    timeout=10,
    )
    todo_id = create_resp.json()["id"]

    resp = requests.get(f"{base_url}/todos/{todo_id}", timeout=10)
    assert resp.status_code == 200
    assert resp.json()["id"] == todo_id

    def test_update_todo(base_url):
    """线上更新待办事项"""
    create_resp = requests.post(
    f"{base_url}/todos",
    json={"title": "更新前", "completed": False},
    timeout=10,
    )
    todo_id = create_resp.json()["id"]

    resp = requests.put(
    f"{base_url}/todos/{todo_id}",
    json={"title": "更新后", "completed": True},
    timeout=10,
    )
    assert resp.status_code == 200
    data = resp.json()
    assert data["title"] == "更新后"
    assert data["completed"] is True

    def test_delete_todo(base_url):
    """线上删除待办事项"""
    create_resp = requests.post(
    f"{base_url}/todos",
    json={"title": "待删除"},
    timeout=10,
    )
    todo_id = create_resp.json()["id"]

    resp = requests.delete(f"{base_url}/todos/{todo_id}", timeout=10)
    assert resp.status_code == 204

    def test_404_for_missing_todo(base_url):
    """查询不存在的待办事项应返回 404"""
    resp = requests.get(f"{base_url}/todos/999999", timeout=10)
    assert resp.status_code == 404
    data = resp.json()
    assert data["code"] == 404

    运行方式:

    # 本地测试
    pytest test_deployed_api.py -v

    # 指定线上地址测试
    DEPLOYED_API_URL="https://your-app.railway.app" pytest test_deployed_api.py -v

    部署测试要点:冒烟测试覆盖健康检查与核心 CRUD 流程,确保线上服务可用;通过环境变量注入线上地址,避免把测试地址硬编码;建议在 CI/CD 流水线中部署完成后自动执行冒烟测试,快速发现回归问题。

    7.5.10 使用 curl 脚本进行端到端冒烟测试

    除了使用 pytest 编写单元测试,我们还可以编写一个轻量级的 shell 脚本,对部署后的 API 进行快速的端到端冒烟测试。这种方式非常适合在 CI/CD 流水线中作为部署后的健康检查步骤。

    #!/usr/bin/env bash
    # smoke_test.sh – API 冒烟测试脚本
    # 用法: ./smoke_test.sh [BASE_URL]
    # 示例: ./smoke_test.sh http://127.0.0.1:8000

    set -e # 任何命令失败则立即退出

    BASE_URL="${1:-http://127.0.0.1:8000}"
    PASS=0
    FAIL=0

    # 辅助函数:检查 HTTP 状态码
    check_status() {
    local expected="$1"
    local actual="$2"
    local description="$3"

    if [ "$expected" -eq "$actual" ]; then
    echo "✅ PASS: $description (HTTP $actual)"
    PASS=$((PASS + 1))
    else
    echo "❌ FAIL: $description (期望 HTTP $expected, 实际 HTTP $actual)"
    FAIL=$((FAIL + 1))
    fi
    }

    echo "========================================"
    echo "开始 API 冒烟测试: $BASE_URL"
    echo "========================================"

    # 1. 健康检查
    echo ""
    echo "— 1. 健康检查 —"
    HEALTH_CODE=$(curl -s -o /dev/null -w "%{http_code}" "$BASE_URL/health")
    check_status 200 "$HEALTH_CODE" "健康检查接口"

    # 2. 创建待办事项
    echo ""
    echo "— 2. 创建待办事项 —"
    CREATE_RESP=$(curl -s -X POST "$BASE_URL/todos" \\
    -H "Content-Type: application/json" \\
    -d '{"title": "冒烟测试任务", "description": "由脚本自动创建"}')

    CREATE_CODE=$(curl -s -o /dev/null -w "%{http_code}" -X POST "$BASE_URL/todos" \\
    -H "Content-Type: application/json" \\
    -d '{"title": "冒烟测试任务", "description": "由脚本自动创建"}')

    check_status 201 "$CREATE_CODE" "创建待办事项"

    # 提取新创建的 ID(使用 Python 解析 JSON)
    TODO_ID=$(echo "$CREATE_RESP" | python3 -c "import sys, json; print(json.load(sys.stdin)['id'])" 2>/dev/null || echo "1")
    echo " 创建成功,新 ID = $TODO_ID"

    # 3. 查询列表
    echo ""
    echo "— 3. 查询待办事项列表 —"
    LIST_CODE=$(curl -s -o /dev/null -w "%{http_code}" "$BASE_URL/todos")
    check_status 200 "$LIST_CODE" "查询待办事项列表"

    # 4. 查询单个
    echo ""
    echo "— 4. 查询单个待办事项 —"
    GET_CODE=$(curl -s -o /dev/null -w "%{http_code}" "$BASE_URL/todos/$TODO_ID")
    check_status 200 "$GET_CODE" "查询单个待办事项 (ID=$TODO_ID)"

    # 5. 更新待办事项
    echo ""
    echo "— 5. 更新待办事项 —"
    UPDATE_CODE=$(curl -s -o /dev/null -w "%{http_code}" -X PUT "$BASE_URL/todos/$TODO_ID" \\
    -H "Content-Type: application/json" \\
    -d '{"title": "冒烟测试任务(已更新)", "completed": true}')

    check_status 200 "$UPDATE_CODE" "更新待办事项 (ID=$TODO_ID)"

    # 6. 删除待办事项
    echo ""
    echo "— 6. 删除待办事项 —"
    DELETE_CODE=$(curl -s -o /dev/null -w "%{http_code}" -X DELETE "$BASE_URL/todos/$TODO_ID")
    check_status 204 "$DELETE_CODE" "删除待办事项 (ID=$TODO_ID)"

    # 7. 验证删除后返回 404
    echo ""
    echo "— 7. 验证删除后返回 404 —"
    NOT_FOUND_CODE=$(curl -s -o /dev/null -w "%{http_code}" "$BASE_URL/todos/$TODO_ID")
    check_status 404 "$NOT_FOUND_CODE" "删除后查询应返回 404"

    echo ""
    echo "========================================"
    echo "测试完成: 通过 $PASS 项, 失败 $FAIL 项"
    echo "========================================"

    # 如果有失败项,以非零状态码退出(供 CI 判断)
    if [ "$FAIL" -gt 0 ]; then
    exit 1
    fi

    运行方式:

    # 给脚本添加执行权限
    chmod +x smoke_test.sh

    # 本地测试
    ./smoke_test.sh http://127.0.0.1:8000

    # 测试部署后的线上环境
    ./smoke_test.sh https://my-api.onrender.com

    冒烟测试要点:脚本覆盖了 CRUD 全流程和健康检查,任何一步失败都会以非零状态码退出;set -e 确保中间步骤失败时立即终止,避免误报;建议将该脚本接入 CI/CD 流水线,在每次部署后自动执行,快速发现回归问题。

    7.5.11 使用 pytest 测试完整 CRUD 流程

    除了针对单个接口的测试,我们还可以编写一个覆盖完整 CRUD 流程的端到端测试,验证「创建 → 查询 → 更新 → 删除」全链路是否正常。下面给出一个完整的测试脚本:

    # test_crud_flow.py
    """
    完整 CRUD 流程测试
    覆盖创建、查询、更新、删除的端到端验证
    """

    import requests
    import pytest

    BASE_URL = "http://127.0.0.1:8000"

    @pytest.fixture()
    def created_todo():
    """创建一条待办事项,测试结束后自动清理"""
    resp = requests.post(
    f"{BASE_URL}/todos",
    json={"title": "端到端测试任务", "description": "验证完整 CRUD 流程"},
    )
    assert resp.status_code == 201
    todo = resp.json()
    yield todo

    # 清理:删除测试数据
    requests.delete(f"{BASE_URL}/todos/{todo['id']}")

    def test_create_todo():
    """创建待办事项应返回 201 和完整数据"""
    resp = requests.post(
    f"{BASE_URL}/todos",
    json={"title": "学习 pytest", "description": "编写自动化测试"},
    )
    assert resp.status_code == 201
    data = resp.json()
    assert data["title"] == "学习 pytest"
    assert data["description"] == "编写自动化测试"
    assert data["completed"] is False
    assert data["id"] is not None
    assert "created_at" in data

    def test_get_todo_list():
    """查询列表应返回数组且包含刚创建的数据"""
    resp = requests.get(f"{BASE_URL}/todos")
    assert resp.status_code == 200
    data = resp.json()
    assert isinstance(data, list)
    assert len(data) > 0

    def test_get_single_todo(created_todo):
    """按 ID 查询单个待办事项"""
    todo_id = created_todo["id"]
    resp = requests.get(f"{BASE_URL}/todos/{todo_id}")
    assert resp.status_code == 200
    data = resp.json()
    assert data["id"] == todo_id
    assert data["title"] == "端到端测试任务"

    def test_update_todo(created_todo):
    """更新待办事项应返回更新后的数据"""
    todo_id = created_todo["id"]
    resp = requests.put(
    f"{BASE_URL}/todos/{todo_id}",
    json={
    "title": "已更新的任务",
    "description": "更新描述",
    "completed": True,
    },
    )
    assert resp.status_code == 200
    data = resp.json()
    assert data["title"] == "已更新的任务"
    assert data["completed"] is True

    def test_delete_todo(created_todo):
    """删除待办事项应返回 204"""
    todo_id = created_todo["id"]
    resp = requests.delete(f"{BASE_URL}/todos/{todo_id}")
    assert resp.status_code == 204

    # 删除后再次查询应返回 404
    resp = requests.get(f"{BASE_URL}/todos/{todo_id}")
    assert resp.status_code == 404

    def test_get_nonexistent_todo():
    """查询不存在的 ID 应返回 404"""
    resp = requests.get(f"{BASE_URL}/todos/999999")
    assert resp.status_code == 404
    data = resp.json()
    assert data["code"] == 404

    运行方式:

    pytest test_crud_flow.py -v

    测试要点:使用 fixture 自动创建和清理测试数据,避免测试之间相互干扰;每个测试只验证一个核心行为,失败时能快速定位问题;建议将 CRUD 流程测试与安全测试、部署测试一起纳入 CI/CD 流水线,形成完整的自动化测试体系。

    7.5.12 使用 pytest 测试完整 CRUD 流程(数据库版
    7.5.13 使用 pytest 测试完整 CRUD 流程(数据库版,完整可运行)

    下面给出一个完整、可直接运行的 pytest 测试脚本,覆盖数据库版 CRUD 的全部接口。它使用 FastAPI 的 TestClient 和临时 SQLite 数据库,确保测试之间互不干扰:

    # test_todo_db.py
    """
    数据库版 CRUD 完整测试
    运行方式:pytest test_todo_db.py -v
    """

    import os
    import tempfile
    import pytest
    from fastapi.testclient import TestClient

    # 使用临时数据库,避免污染开发数据
    TEMP_DB = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
    os.environ["DATABASE_URL"] = f"sqlite:///{TEMP_DB.name}"

    # 注意:必须在导入 main 之前设置好环境变量
    from main import app, Base, engine, SessionLocal # noqa: E402

    # 重建表结构
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)

    client = TestClient(app)

    @pytest.fixture(autouse=True)
    def clean_db():
    """每个测试用例前清空数据表,保证用例独立"""
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)
    yield
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)

    def test_create_todo():
    """测试创建待办事项"""
    resp = client.post("/todos", json={"title": "学习 FastAPI", "description": "完成教程"})
    assert resp.status_code == 201
    data = resp.json()
    assert data["title"] == "学习 FastAPI"
    assert data["completed"] is False
    assert data["id"] == 1

    def test_list_todos_empty():
    """测试空列表"""
    resp = client.get("/todos")
    assert resp.status_code == 200
    assert resp.json() == []

    def test_list_todos_with_filter():
    """测试按完成状态筛选"""
    client.post("/todos", json={"title": "任务A", "completed": False})
    client.post("/todos", json={"title": "任务B", "completed": True})

    resp = client.get("/todos", params={"completed": True})
    assert resp.status_code == 200
    data = resp.json()
    assert len(data) == 1
    assert data[0]["title"] == "任务B"

    def test_get_todo_success():
    """测试查询单个待办事项"""
    created = client.post("/todos", json={"title": "任务A"}).json()
    resp = client.get(f"/todos/{created['id']}")
    assert resp.status_code == 200
    assert resp.json()["title"] == "任务A"

    def test_get_todo_not_found():
    """测试查询不存在的待办事项"""
    resp = client.get("/todos/999")
    assert resp.status_code == 404
    assert resp.json()["code"] == 404

    def test_update_todo():
    """测试更新待办事项"""
    created = client.post("/todos", json={"title": "任务A"}).json()
    resp = client.put(
    f"/todos/{created['id']}",
    json={"title": "任务A(已更新)", "description": "新描述", "completed": True},
    )
    assert resp.status_code == 200
    data = resp.json()
    assert data["title"] == "任务A(已更新)"
    assert data["completed"] is True

    def test_update_todo_not_found():
    """测试更新不存在的待办事项"""
    resp = client.put("/todos/999", json={"title": "不存在"})
    assert resp.status_code == 404

    def test_delete_todo():
    """测试删除待办事项"""
    created = client.post("/todos", json={"title": "待删除"}).json()
    resp = client.delete(f"/todos/{created['id']}")
    assert resp.status_code == 204

    # 删除后再次查询应返回 404
    resp = client.get(f"/todos/{created['id']}")
    assert resp.status_code == 404

    def test_delete_todo_not_found():
    """测试删除不存在的待办事项"""
    resp = client.delete("/todos/999")
    assert resp.status_code == 404

    def test_validation_error():
    """测试缺少必填字段时返回 422"""
    resp = client.post("/todos", json={"description": "缺少 title"})
    assert resp.status_code == 422
    data = resp.json()
    assert data["code"] == 422
    assert data["data"][0]["field"] == "title"

    def test_full_crud_flow():
    """测试完整 CRUD 流程"""
    # 1. 创建
    resp = client.post("/todos", json={"title": "完整流程", "description": "测试"})
    assert resp.status_code == 201
    todo_id = resp.json()["id"]

    # 2. 查询列表
    resp = client.get("/todos")
    assert resp.status_code == 200
    assert len(resp.json()) == 1

    # 3. 查询单个
    resp = client.get(f"/todos/{todo_id}")
    assert resp.status_code == 200

    # 4. 更新
    resp = client.put(f"/todos/{todo_id}", json={"title": "完整流程(完成)", "completed": True})
    assert resp.status_code == 200
    assert resp.json()["completed"] is True

    # 5. 删除
    resp = client.delete(f"/todos/{todo_id}")
    assert resp.status_code == 204

    # 6. 确认已删除
    resp = client.get(f"/todos/{todo_id}")
    assert resp.status_code == 404

    运行方式:

    # 安装测试依赖
    pip install pytest httpx

    # 运行全部测试
    pytest test_todo_db.py -v

    测试要点:每个用例通过 clean_db fixture 重建数据表,保证用例之间完全隔离;TestClient 直接调用应用,无需启动真实服务器;覆盖了成功、失败、参数校验、完整流程等 12 个典型场景,可作为回归测试长期保留。

    前面我们分别测试了内存版和数据库版的接口。下面给出一个针对数据库版 CRUD 的完整 pytest 测试脚本,覆盖创建、查询、更新、删除全流程,并验证数据确实持久化到 SQLite:

    # test_crud_db.py
    """
    数据库版 CRUD 完整测试
    覆盖:创建、查询列表、查询单个、更新、删除、持久化验证
    """

    import os
    import pytest
    from fastapi.testclient import TestClient

    # 使用临时数据库,避免污染开发数据
    os.environ["DATABASE_URL"] = "sqlite:///./test_todos.db"

    from main import app, Base, engine, SessionLocal, TodoDB

    client = TestClient(app)

    @pytest.fixture(scope="module", autouse=True)
    def setup_database():
    """每个测试模块开始时重建数据库表"""
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)
    yield
    # 测试结束后清理临时数据库文件
    Base.metadata.drop_all(bind=engine)
    if os.path.exists("./test_todos.db"):
    os.remove("./test_todos.db")

    def test_create_todo():
    """创建待办事项应返回 201 和完整数据"""
    resp = client.post("/todos", json={
    "title": "学习 FastAPI",
    "description": "完成数据库版 CRUD 测试",
    })
    assert resp.status_code == 201
    data = resp.json()
    assert data["title"] == "学习 FastAPI"
    assert data["completed"] is False
    assert data["id"] == 1
    return data

    def test_list_todos():
    """查询列表应包含刚创建的数据"""
    resp = client.get("/todos")
    assert resp.status_code == 200
    data = resp.json()
    assert len(data) >= 1
    assert any(t["title"] == "学习 FastAPI" for t in data)

    def test_get_single_todo():
    """按 ID 查询单个待办事项"""
    resp = client.get("/todos/1")
    assert resp.status_code == 200
    data = resp.json()
    assert data["id"] == 1
    assert data["title"] == "学习 FastAPI"

    def test_get_nonexistent_todo():
    """查询不存在的 ID 应返回 404"""
    resp = client.get("/todos/999")
    assert resp.status_code == 404
    assert resp.json()["code"] == 404

    def test_update_todo():
    """更新待办事项应返回更新后的数据"""
    resp = client.put("/todos/1", json={
    "title": "学习 FastAPI(已完成)",
    "description": "数据库版 CRUD 测试通过",
    "completed": True,
    })
    assert resp.status_code == 200
    data = resp.json()
    assert data["title"] == "学习 FastAPI(已完成)"
    assert data["completed"] is True

    def test_delete_todo():
    """删除待办事项应返回 204"""
    resp = client.delete("/todos/1")
    assert resp.status_code == 204

    def test_delete_nonexistent_todo():
    """删除不存在的 ID 应返回 404"""
    resp = client.delete("/todos/999")
    assert resp.status_code == 404

    def test_data_persisted_in_db():
    """验证数据确实写入 SQLite 数据库文件"""
    # 直接查询数据库,绕过 API
    db = SessionLocal()
    try:
    # 先创建一条数据
    client.post("/todos", json={"title": "持久化验证", "description": "直接查库"})
    # 从数据库读取
    todo = db.query(TodoDB).filter(TodoDB.title == "持久化验证").first()
    assert todo is not None
    assert todo.description == "直接查库"
    finally:
    db.close()

    运行方式:

    pytest test_crud_db.py -v

    数据库版测试要点:使用 TestClient 模拟真实 HTTP 请求,无需启动服务器;通过 Base.metadata.drop_all/create_all 在测试前后重建表,保证测试隔离;最后一条用例直接查询数据库,验证数据确实落盘,而不仅仅是内存操作。

    7.5.14 使用 pytest 测试完整 CRUD 流程(数据库版,含错误场景
    7.5.15 使用 pytest 测试完整 CRUD 流程(数据库版,含错误场景)

    在 7.5.14 的基础上,补充一个完整的端到端测试脚本,覆盖数据库版 CRUD 的全部成功与错误场景,并加入并发与边界测试,确保接口在生产环境下的健壮性。

    # test_todo_e2e.py
    """
    待办事项 API 端到端测试(数据库版)
    覆盖:CRUD 成功路径、404/422 错误路径、并发写入、边界值、数据持久化
    """

    import threading
    import time
    import pytest
    from fastapi.testclient import TestClient
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.pool import StaticPool

    from main import app, Base, get_db, TodoDB

    # 使用独立的内存数据库,避免污染开发数据
    engine = create_engine(
    "sqlite://",
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
    )
    TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

    def override_get_db():
    db = TestingSessionLocal()
    try:
    yield db
    finally:
    db.close()

    app.dependency_overrides[get_db] = override_get_db
    client = TestClient(app)

    @pytest.fixture(autouse=True)
    def setup_db():
    """每个测试前重建数据表,保证用例相互独立"""
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)
    yield

    def create_todo(title="学习 FastAPI", description="完成教程", completed=False):
    """辅助函数:创建待办事项"""
    return client.post(
    "/todos",
    json={"title": title, "description": description, "completed": completed},
    )

    # ========== 创建接口测试 ==========
    def test_create_todo_success():
    """正常创建返回 201 且包含自增 ID"""
    resp = create_todo()
    assert resp.status_code == 201
    data = resp.json()
    assert data["id"] == 1
    assert data["title"] == "学习 FastAPI"
    assert data["completed"] is False
    assert "created_at" in data

    def test_create_todo_missing_title():
    """缺少必填字段 title 返回 422"""
    resp = client.post("/todos", json={"description": "没有标题"})
    assert resp.status_code == 422
    body = resp.json()
    assert body["code"] == 422
    assert "title" in str(body["data"])

    def test_create_todo_wrong_type():
    """title 类型错误返回 422"""
    resp = client.post("/todos", json={"title": 12345})
    assert resp.status_code == 422

    def test_create_todo_empty_title():
    """空字符串 title 返回 422(可结合 Pydantic 校验器)"""
    resp = client.post("/todos", json={"title": ""})
    assert resp.status_code in (201, 422) # 取决于是否配置 min_length

    # ========== 查询接口测试 ==========
    def test_list_todos_empty():
    """空数据库返回空列表"""
    resp = client.get("/todos")
    assert resp.status_code == 200
    assert resp.json() == []

    def test_list_todos_with_filter():
    """按 completed 筛选"""
    create_todo("任务A", completed=False)
    create_todo("任务B", completed=True)

    resp = client.get("/todos", params={"completed": True})
    assert resp.status_code == 200
    data = resp.json()
    assert len(data) == 1
    assert data[0]["title"] == "任务B"

    def test_get_todo_success():
    """查询单个待办事项"""
    created = create_todo().json()
    resp = client.get(f"/todos/{created['id']}")
    assert resp.status_code == 200
    assert resp.json()["title"] == "学习 FastAPI"

    def test_get_todo_not_found():
    """查询不存在的 ID 返回 404"""
    resp = client.get("/todos/999")
    assert resp.status_code == 404
    assert resp.json()["code"] == 404

    def test_get_todo_invalid_id():
    """非数字 ID 返回 422"""
    resp = client.get("/todos/abc")
    assert resp.status_code == 422

    # ========== 更新接口测试 ==========
    def test_update_todo_success():
    """正常更新字段"""
    created = create_todo().json()
    resp = client.put(
    f"/todos/{created['id']}",
    json={"title": "已更新", "description": "新描述", "completed": True},
    )
    assert resp.status_code == 200
    data = resp.json()
    assert data["title"] == "已更新"
    assert data["completed"] is True
    assert data["id"] == created["id"] # ID 不变

    def test_update_todo_not_found():
    """更新不存在的 ID 返回 404"""
    resp = client.put(
    "/todos/999",
    json={"title": "不存在", "description": "", "completed": False},
    )
    assert resp.status_code == 404

    def test_update_todo_partial_fields():
    """只更新部分字段(description 为 None 时保留原值)"""
    created = create_todo(description="原始描述").json()
    resp = client.put(
    f"/todos/{created['id']}",
    json={"title": "只改标题", "description": None, "completed": False},
    )
    assert resp.status_code == 200
    assert resp.json()["title"] == "只改标题"

    # ========== 删除接口测试 ==========
    def test_delete_todo_success():
    """正常删除返回 204"""
    created = create_todo().json()
    resp = client.delete(f"/todos/{created['id']}")
    assert resp.status_code == 204

    def test_delete_todo_not_found():
    """删除不存在的 ID 返回 404"""
    resp = client.delete("/todos/999")
    assert resp.status_code == 404

    def test_delete_then_get_returns_404():
    """删除后再查询返回 404"""
    created = create_todo().json()
    client.delete(f"/todos/{created['id']}")
    resp = client.get(f"/todos/{created['id']}")
    assert resp.status_code == 404

    # ========== 边界与并发测试 ==========
    def test_create_many_todos():
    """批量创建 100 条,验证 ID 自增与列表长度"""
    for i in range(100):
    resp = create_todo(title=f"任务{i}")
    assert resp.status_code == 201

    resp = client.get("/todos")
    assert len(resp.json()) == 100
    assert resp.json()[0]["id"] == 1
    assert resp.json()[1]["id"] == 100

    def test_concurrent_create():
    """并发创建 20 条,验证无 ID 冲突"""
    results = []

    def worker(n):
    resp = create_todo(title=f"并发任务{n}")
    results.append(resp.status_code)

    threads = [threading.Thread(target=worker, args=(i,)) for i in range(20)]
    for t in threads:
    t.start()
    for t in threads:
    t.join()

    assert all(code == 201 for code in results)
    resp = client.get("/todos")
    assert len(resp.json()) == 20

    def test_data_persisted_in_db():
    """创建后数据确实写入数据库"""
    create_todo(title="持久化验证")
    db = TestingSessionLocal()
    todo = db.query(TodoDB).filter(TodoDB.title == "持久化验证").first()
    db.close()
    assert todo is not None
    assert todo.completed is False

    运行方式:

    # 安装依赖
    pip install fastapi uvicorn sqlalchemy pytest httpx

    # 运行端到端测试
    pytest test_todo_e2e.py -v

    测试要点:覆盖创建(成功/缺字段/类型错误/空标题)、查询(空列表/筛选/成功/404/422)、更新(成功/404/部分字段)、删除(成功/404/删除后查询)、边界(批量 100 条/并发 20 线程/数据库持久化)共 5 大类 18 个用例;每个用例独立重建内存数据库,互不干扰;并发测试验证了 SQLite 在 check_same_thread=False 配置下的写入安全性。

    在 7.5.13 的基础上,补充覆盖错误场景的测试用例,包括:资源不存在(404)、参数校验失败(422)、重复创建、非法 ID 等,确保接口在异常情况下也能返回正确的状态码和错误信息。

    # test_crud_errors.py
    """
    数据库版 CRUD 错误场景测试
    覆盖:404 资源不存在、422 参数校验失败、非法 ID、重复创建等
    """

    import pytest
    from fastapi.testclient import TestClient
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.pool import StaticPool

    from main import app, Base, get_db, TodoDB

    # 使用内存数据库,每个测试独立重建
    engine = create_engine(
    "sqlite://",
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
    )
    TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

    def override_get_db():
    """测试用数据库会话"""
    db = TestingSessionLocal()
    try:
    yield db
    finally:
    db.close()

    app.dependency_overrides[get_db] = override_get_db
    client = TestClient(app)

    @pytest.fixture(autouse=True)
    def setup_db():
    """每个测试前重建数据表"""
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)
    yield

    def create_todo(title="测试待办", description="测试描述"):
    """辅助函数:创建一条待办"""
    return client.post(
    "/todos",
    json={"title": title, "description": description},
    )

    # ========== 404 资源不存在 ==========
    def test_get_nonexistent_todo():
    """查询不存在的待办返回 404"""
    resp = client.get("/todos/999")
    assert resp.status_code == 404
    assert resp.json()["detail"] == "待办事项不存在"

    def test_update_nonexistent_todo():
    """更新不存在的待办返回 404"""
    resp = client.put(
    "/todos/999",
    json={"title": "不存在", "description": "无", "completed": False},
    )
    assert resp.status_code == 404

    def test_delete_nonexistent_todo():
    """删除不存在的待办返回 404"""
    resp = client.delete("/todos/999")
    assert resp.status_code == 404

    # ========== 422 参数校验失败 ==========
    def test_create_missing_title():
    """缺少必填字段 title 返回 422"""
    resp = client.post("/todos", json={"description": "缺少标题"})
    assert resp.status_code == 422
    data = resp.json()
    assert data["code"] == 422
    assert "title" in str(data["data"])

    def test_create_wrong_type():
    """completed 传字符串而非布尔值返回 422"""
    resp = client.post(
    "/todos",
    json={"title": "类型错误", "completed": "true"},
    )
    assert resp.status_code == 422

    def test_get_invalid_id_type():
    """路径参数非整数返回 422"""
    resp = client.get("/todos/abc")
    assert resp.status_code == 422

    # ========== 正常流程 + 边界 ==========
    def test_create_and_get_flow():
    """创建后能正确查询到"""
    create_resp = create_todo("边界测试")
    assert create_resp.status_code == 201
    todo_id = create_resp.json()["id"]

    get_resp = client.get(f"/todos/{todo_id}")
    assert get_resp.status_code == 200
    assert get_resp.json()["title"] == "边界测试"

    def test_create_multiple_todos():
    """连续创建多条,ID 自增且互不干扰"""
    ids = []
    for i in range(3):
    resp = create_todo(f"待办 {i}")
    assert resp.status_code == 201
    ids.append(resp.json()["id"])

    assert ids == [1, 2, 3]

    list_resp = client.get("/todos")
    assert len(list_resp.json()) == 3

    def test_filter_by_completed():
    """按完成状态筛选"""
    create_todo("未完成")
    resp = client.post(
    "/todos",
    json={"title": "已完成", "completed": True},
    )
    assert resp.status_code == 201

    pending = client.get("/todos?completed=false")
    done = client.get("/todos?completed=true")

    assert len(pending.json()) == 1
    assert len(done.json()) == 1
    assert done.json()[0]["title"] == "已完成"

    def test_update_preserves_id_and_created_at():
    """更新后 ID 和创建时间保持不变"""
    create_resp = create_todo("原始标题")
    todo_id = create_resp.json()["id"]
    original_created_at = create_resp.json()["created_at"]

    update_resp = client.put(
    f"/todos/{todo_id}",
    json={"title": "新标题", "description": "新描述", "completed": True},
    )
    assert update_resp.status_code == 200
    data = update_resp.json()
    assert data["id"] == todo_id
    assert data["created_at"] == original_created_at
    assert data["title"] == "新标题"
    assert data["completed"] is True

    def test_delete_then_get_returns_404():
    """删除后再查询返回 404"""
    create_resp = create_todo("待删除")
    todo_id = create_resp.json()["id"]

    delete_resp = client.delete(f"/todos/{todo_id}")
    assert delete_resp.status_code == 204

    get_resp = client.get(f"/todos/{todo_id}")
    assert get_resp.status_code == 404

    def test_empty_list():
    """空数据库返回空列表"""
    resp = client.get("/todos")
    assert resp.status_code == 200
    assert resp.json() == []

    运行方式:

    # 安装测试依赖
    pip install pytest httpx

    # 运行错误场景测试
    pytest test_crud_errors.py -v

    测试要点:覆盖 404(查询/更新/删除不存在资源)、422(缺字段/类型错误/非法路径参数)、正常流程(创建查询、批量创建、状态筛选、更新保留 ID 与创建时间、删除后 404)、空列表共 4 大类 12 个用例;每个用例独立重建内存数据库,互不干扰;app.dependency_overrides 将数据库会话替换为测试内存库,无需真实 SQLite 文件。

    7.5.16 使用 pytest 测试内存版 CRUD 接口(完整可运行)

    下面提供一个针对内存版 CRUD 接口的完整 pytest 测试脚本。它使用 FastAPI 自带的 TestClient,无需启动真实服务器即可运行,适合在开发阶段快速验证接口逻辑。

    # test_todo_memory.py
    """
    内存版 CRUD 接口测试
    覆盖:创建、列表、查询单个、更新、删除、错误处理
    """

    import pytest
    from fastapi.testclient import TestClient

    # 假设内存版接口定义在 main_memory.py 中
    from main_memory import app

    client = TestClient(app)

    @pytest.fixture(autouse=True)
    def reset_todos():
    """每个测试前重置内存数据"""
    from main_memory import todos, todo_id_counter
    todos.clear()
    todo_id_counter = 1
    yield

    def create_todo(title="学习 FastAPI", description="完成 API 开发教程"):
    """辅助函数:创建待办事项"""
    return client.post("/todos", json={"title": title, "description": description})

    # ========== 创建 ==========
    def test_create_todo_success():
    """正常创建返回 201 和待办事项"""
    resp = create_todo()
    assert resp.status_code == 201
    data = resp.json()
    assert data["id"] == 1
    assert data["title"] == "学习 FastAPI"
    assert data["completed"] is False

    def test_create_todo_missing_title():
    """缺少必填字段 title 返回 422"""
    resp = client.post("/todos", json={"description": "没有标题"})
    assert resp.status_code == 422

    def test_create_todo_auto_increment_id():
    """连续创建 ID 自动递增"""
    create_todo("任务一")
    resp = create_todo("任务二")
    assert resp.json()["id"] == 2

    # ========== 查询列表 ==========
    def test_list_todos_empty():
    """空列表返回 []"""
    resp = client.get("/todos")
    assert resp.status_code == 200
    assert resp.json() == []

    def test_list_todos_with_data():
    """有数据时返回全部"""
    create_todo("任务一")
    create_todo("任务二")
    resp = client.get("/todos")
    assert resp.status_code == 200
    assert len(resp.json()) == 2

    def test_list_todos_filter_completed():
    """按完成状态筛选"""
    create_todo("未完成")
    resp = client.post("/todos", json={"title": "已完成", "completed": True})
    assert resp.status_code == 201

    resp = client.get("/todos", params={"completed": True})
    data = resp.json()
    assert len(data) == 1
    assert data[0]["title"] == "已完成"

    # ========== 查询单个 ==========
    def test_get_todo_success():
    """查询存在的待办事项"""
    create_todo()
    resp = client.get("/todos/1")
    assert resp.status_code == 200
    assert resp.json()["title"] == "学习 FastAPI"

    def test_get_todo_not_found():
    """查询不存在的 ID 返回 404"""
    resp = client.get("/todos/999")
    assert resp.status_code == 404
    assert resp.json()["detail"] == "待办事项不存在"

    # ========== 更新 ==========
    def test_update_todo_success():
    """正常更新返回更新后的数据"""
    create_todo()
    resp = client.put(
    "/todos/1",
    json={"title": "学习 FastAPI", "description": "已完成", "completed": True},
    )
    assert resp.status_code == 200
    data = resp.json()
    assert data["completed"] is True
    assert data["description"] == "已完成"

    def test_update_todo_not_found():
    """更新不存在的 ID 返回 404"""
    resp = client.put(
    "/todos/999",
    json={"title": "不存在", "description": "", "completed": False},
    )
    assert resp.status_code == 404

    # ========== 删除 ==========
    def test_delete_todo_success():
    """正常删除返回 204"""
    create_todo()
    resp = client.delete("/todos/1")
    assert resp.status_code == 204

    def test_delete_todo_not_found():
    """删除不存在的 ID 返回 404"""
    resp = client.delete("/todos/999")
    assert resp.status_code == 404

    def test_delete_todo_removes_data():
    """删除后列表不再包含该数据"""
    create_todo()
    client.delete("/todos/1")
    resp = client.get("/todos")
    assert resp.json() == []

    运行方式:

    # 安装测试依赖
    pip install pytest httpx

    # 运行测试
    pytest test_todo_memory.py -v

    测试要点:覆盖创建(成功/缺字段/ID 自增)、列表(空/有数据/筛选)、查询单个(成功/404)、更新(成功/404)、删除(成功/404/数据移除)共 5 大类 13 个用例;每个用例通过 fixture 重置内存数据,保证用例之间相互独立;TestClient 直接调用应用,无需启动真实端口,测试速度更快。

    8. 部署上线

    接口开发、测试完成后,下一步就是把它部署到公网,让其他用户能够访问。本章介绍从 Docker 容器化到云平台部署的完整流程。

    8.1 使用 Docker 部署

    Docker 可以将应用及其依赖打包成镜像,实现「一次构建,到处运行」。首先创建 Dockerfile:

    # 使用 Python 3.11 作为基础镜像
    FROM python:3.11-slim

    # 设置工作目录
    WORKDIR /app

    # 复制依赖文件并安装
    COPY requirements.txt .
    RUN pip install –no-cache-dir -r requirements.txt

    # 复制项目代码
    COPY . .

    # 暴露端口
    EXPOSE 8000

    # 启动命令
    CMD ["uvicorn", "main:app", "–host", "0.0.0.0", "–port", "8000"]

    创建 requirements.txt:

    fastapi==0.115.0
    uvicorn[standard]==0.30.0
    sqlalchemy==2.0.30
    pydantic==2.7.0

    构建并运行镜像:

    # 构建镜像
    docker build -t todo-api .

    # 运行容器
    docker run -d –name todo-api -p 8000:8000 todo-api

    # 查看日志
    docker logs -f todo-api

    8.2 部署到云平台

    除了自建服务器,还可以使用云平台快速部署,省去服务器运维的麻烦。

    8.2.1 使用 Railway / Render 部署(推荐新手)

    Railway 和 Render 支持直接从 GitHub 仓库部署,无需手动配置服务器:

  • 将代码推送到 GitHub 仓库
  • 在 Railway / Render 后台选择「New Project」→「Deploy from GitHub」
  • 选择仓库,平台会自动识别 Python 项目并安装依赖
  • 设置环境变量(如 DATABASE_URL、SECRET_KEY)
  • 点击 Deploy,等待构建完成即可获得公网 URL
  • # 推送代码到 GitHub
    git init
    git add .
    git commit -m "初始化待办事项 API"
    git remote add origin https://github.com/yourname/todo-api.git
    git push -u origin main

    8.2.2 使用云服务器 + Docker 部署(阿里云 / 腾讯云)

    如果你已经有一台云服务器(如阿里云 ECS、腾讯云 CVM),可以这样部署:

    # 1. 在服务器上安装 Docker
    curl -fsSL https://get.docker.com | sh

    # 2. 将代码上传到服务器(使用 scp 或 git clone)
    git clone https://github.com/yourname/todo-api.git
    cd todo-api

    # 3. 构建并运行容器
    docker build -t todo-api .
    docker run -d –name todo-api -p 8000:8000 todo-api

    # 4. 配置 Nginx 反向代理(可选)
    # 在 /etc/nginx/sites-available/todo-api 中配置:
    # server {
    # listen 80;
    # server_name api.example.com;
    # location / {
    # proxy_pass http://127.0.0.1:8000;
    # proxy_set_header Host $host;
    # }
    # }

    8.2.3 使用 Serverless 平台部署(按调用量计费)

    Serverless 平台(如 AWS Lambda、腾讯云 SCF)按调用次数计费,适合流量波动大的场景。以腾讯云 SCF 为例:

    # serverless.py
    from mangum import Mangum
    from main import app

    # 将 FastAPI 应用包装为 Serverless 函数
    handler = Mangum(app)

    # serverless.yml
    service: todoapi

    provider:
    name: tencent
    runtime: python3.11

    functions:
    api:
    handler: serverless.handler
    events:
    http:
    path: /{proxy+}
    method: any

    8.2.4 部署注意事项

    部署到生产环境时,需要注意以下几点:

    • 环境变量管理:数据库连接串、JWT 密钥等敏感信息通过环境变量注入,不要硬编码在代码中
    • 数据库迁移:使用 Alembic 管理数据库结构变更,避免手动改表
    • HTTPS 证书:通过 Nginx 或云平台配置 SSL 证书,强制 HTTPS
    • 日志持久化:将容器日志输出到文件或日志平台,便于排查问题
    • 自动重启:配置进程守护(如 systemd、supervisor),服务崩溃后自动拉起
    8.2.5 部署上线检查清单

    上线前,建议逐项确认以下事项:

    • 所有敏感信息已通过环境变量注入,未硬编码
    • 数据库已配置迁移脚本,数据可持久化
    • HTTPS 证书已配置,无明文 HTTP 访问
    • 日志已接入集中管理平台
    • 健康检查接口已就绪
    • 限流策略已配置,防止滥用
    • 依赖已通过安全扫描
    8.2.6 部署上线完整流程示例

    下面给出一个从代码到上线的完整流程示例,涵盖构建、推送、部署和验证:

    # 1. 构建 Docker 镜像
    docker build -t my-api:latest .

    # 2. 本地验证镜像可正常运行
    docker run -d -p 8000:8000 –name my-api-test my-api:latest
    curl http://127.0.0.1:8000/health

    # 3. 登录容器镜像仓库(以 Docker Hub 为例)
    docker login
    docker tag my-api:latest yourusername/my-api:latest
    docker push yourusername/my-api:latest

    # 4. 在云服务器上拉取并运行
    ssh root@your-server-ip
    docker pull yourusername/my-api:latest
    docker run -d -p 8000:8000 \\
    –name my-api \\
    –restart always \\
    -e DATABASE_URL="sqlite:///./todos.db" \\
    -e SECRET_KEY="your-secret-key" \\
    -v /data/todos.db:/app/todos.db \\
    yourusername/my-api:latest

    # 5. 验证部署结果
    curl http://your-server-ip:8000/health
    curl http://your-server-ip:8000/todos

    部署流程说明

    • 构建阶段:使用 Dockerfile 将应用打包成镜像,确保构建产物可复现
    • 推送阶段:将镜像推送到仓库,便于版本管理和回滚
    • 部署阶段:在服务器上拉取镜像并运行,通过 -e 注入环境变量,通过 -v 挂载数据卷持久化数据
    • 验证阶段:通过健康检查接口确认服务正常,再验证业务接口

    部署建议:生产环境建议使用 docker-compose 或 Kubernetes 管理多容器编排;数据库使用外部托管服务(如云数据库),避免数据与容器生命周期绑定。

    8.2.9 部署步骤截图说明

    下面以 Railway 平台为例,展示部署的关键步骤。由于截图无法在本文中直接展示,这里用文字 + 代码块模拟部署过程中的关键界面和操作指引,你可以对照实际操作截图保存。

    步骤 1:创建新项目

    登录 Railway 后,点击「New Project」按钮,选择「Deploy from GitHub repo」,关联你的代码仓库:

    ┌─────────────────────────────────────────────┐
    │ Railway Dashboard │
    │ │
    │ [+ New Project] │
    │ ┌───────────────────────────────────────┐ │
    │ │ Deploy from GitHub repo │ │
    │ │ Deploy from Dockerfile │ │
    │ │ Deploy from template │ │
    │ └───────────────────────────────────────┘ │
    └─────────────────────────────────────────────┘

    步骤 2:配置启动命令

    在项目 Settings 中找到「Deploy」配置,设置启动命令:

    ┌─────────────────────────────────────────────┐
    │ Deploy Settings │
    │ │
    │ Root Directory: / │
    │ Build Command: pip install -r requirements.txt │
    │ Start Command: uvicorn main:app –host 0.0.0.0 –port $PORT │
    │ │
    │ [Deploy] │
    └─────────────────────────────────────────────┘

    步骤 3:添加环境变量

    在「Variables」选项卡中添加 SECRET_KEY 等环境变量:

    ┌─────────────────────────────────────────────┐
    │ Variables │
    │ │
    │ SECRET_KEY = your-secret-key │
    │ DATABASE_URL = postgresql://… │
    │ │
    │ [+ New Variable] │
    └─────────────────────────────────────────────┘

    步骤 4:查看部署日志

    部署过程中,点击「Deployments」查看实时日志,确认服务启动成功:

    ┌─────────────────────────────────────────────┐
    │ Deploy Logs │
    │ │
    │ ✓ Collecting fastapi… │
    │ ✓ Installing dependencies… │
    │ ✓ Uvicorn running on http://0.0.0.0:8000 │
    │ ✓ Application startup complete. │
    │ │
    │ [View Live Logs] │
    └─────────────────────────────────────────────┘

    步骤 5:访问线上服务

    部署成功后,Railway 会分配一个公网域名,如 https://my-api-production.up.railway.app。在浏览器中访问该域名,即可看到 API 返回的 JSON 数据。

    截图建议:实际部署时,建议按上述 5 个步骤分别截图,保存为 deploy-step1.png ~ deploy-step5.png,插入到对应步骤下方,方便读者对照操作。

    8.2.10 部署流程时序图

    为了更直观地理解从代码提交到服务上线的完整流程,下面用一张时序图展示典型的 CI/CD 部署流水线:

    云平台

    (Railway/Render/云服务器)

    镜像仓库

    (Docker Hub)

    CI 流水线

    (GitHub Actions)

    Git 仓库

    (GitHub/Gitee)

    开发者

    云平台

    (Railway/Render/云服务器)

    镜像仓库

    (Docker Hub)

    CI 流水线

    (GitHub Actions)

    Git 仓库

    (GitHub/Gitee)

    开发者

    #mermaid-svg-y2YawUujOSEjNnRs{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-y2YawUujOSEjNnRs .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-y2YawUujOSEjNnRs .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-y2YawUujOSEjNnRs .error-icon{fill:#552222;}#mermaid-svg-y2YawUujOSEjNnRs .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-y2YawUujOSEjNnRs .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-y2YawUujOSEjNnRs .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-y2YawUujOSEjNnRs .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-y2YawUujOSEjNnRs .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-y2YawUujOSEjNnRs .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-y2YawUujOSEjNnRs .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-y2YawUujOSEjNnRs .marker{fill:#333333;stroke:#333333;}#mermaid-svg-y2YawUujOSEjNnRs .marker.cross{stroke:#333333;}#mermaid-svg-y2YawUujOSEjNnRs svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-y2YawUujOSEjNnRs p{margin:0;}#mermaid-svg-y2YawUujOSEjNnRs .actor{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-y2YawUujOSEjNnRs text.actor>tspan{fill:black;stroke:none;}#mermaid-svg-y2YawUujOSEjNnRs .actor-line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-y2YawUujOSEjNnRs .innerArc{stroke-width:1.5;stroke-dasharray:none;}#mermaid-svg-y2YawUujOSEjNnRs .messageLine0{stroke-width:1.5;stroke-dasharray:none;stroke:#333;}#mermaid-svg-y2YawUujOSEjNnRs .messageLine1{stroke-width:1.5;stroke-dasharray:2,2;stroke:#333;}#mermaid-svg-y2YawUujOSEjNnRs #arrowhead path{fill:#333;stroke:#333;}#mermaid-svg-y2YawUujOSEjNnRs .sequenceNumber{fill:white;}#mermaid-svg-y2YawUujOSEjNnRs #sequencenumber{fill:#333;}#mermaid-svg-y2YawUujOSEjNnRs #crosshead path{fill:#333;stroke:#333;}#mermaid-svg-y2YawUujOSEjNnRs .messageText{fill:#333;stroke:none;}#mermaid-svg-y2YawUujOSEjNnRs .labelBox{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-y2YawUujOSEjNnRs .labelText,#mermaid-svg-y2YawUujOSEjNnRs .labelText>tspan{fill:black;stroke:none;}#mermaid-svg-y2YawUujOSEjNnRs .loopText,#mermaid-svg-y2YawUujOSEjNnRs .loopText>tspan{fill:black;stroke:none;}#mermaid-svg-y2YawUujOSEjNnRs .loopLine{stroke-width:2px;stroke-dasharray:2,2;stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-y2YawUujOSEjNnRs .note{stroke:#aaaa33;fill:#fff5ad;}#mermaid-svg-y2YawUujOSEjNnRs .noteText,#mermaid-svg-y2YawUujOSEjNnRs .noteText>tspan{fill:black;stroke:none;}#mermaid-svg-y2YawUujOSEjNnRs .activation0{fill:#f4f4f4;stroke:#666;}#mermaid-svg-y2YawUujOSEjNnRs .activation1{fill:#f4f4f4;stroke:#666;}#mermaid-svg-y2YawUujOSEjNnRs .activation2{fill:#f4f4f4;stroke:#666;}#mermaid-svg-y2YawUujOSEjNnRs .actorPopupMenu{position:absolute;}#mermaid-svg-y2YawUujOSEjNnRs .actorPopupMenuPanel{position:absolute;fill:#ECECFF;box-shadow:0px 8px 16px 0px rgba(0,0,0,0.2);filter:drop-shadow(3px 5px 2px rgb(0 0 0 / 0.4));}#mermaid-svg-y2YawUujOSEjNnRs .actor-man line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-y2YawUujOSEjNnRs .actor-man circle,#mermaid-svg-y2YawUujOSEjNnRs line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;stroke-width:2px;}#mermaid-svg-y2YawUujOSEjNnRs :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

    1. git push 提交代码

    2. 触发 CI 流水线

    3. 运行自动化测试

    (pytest)

    4. 构建并推送 Docker 镜像

    5. 拉取最新镜像

    6. 启动容器并执行健康检查

    7. 返回部署状态与访问地址

    对应的部署步骤说明如下:

    步骤操作说明
    1 代码提交 开发者将代码推送到 Git 仓库(如 GitHub、Gitee)
    2 触发流水线 推送事件自动触发 CI 流水线(如 GitHub Actions、GitLab CI)
    3 自动化测试 流水线运行 pytest 等测试,确保代码质量
    4 构建镜像 测试通过后构建 Docker 镜像并推送到镜像仓库
    5 拉取镜像 云平台检测到新镜像后自动拉取
    6 启动服务 云平台启动新容器,并执行健康检查确认服务可用
    7 部署完成 部署成功后返回访问地址,开发者可进行验证

    部署流程要点:将测试、构建、部署全部自动化,可以显著降低人工操作带来的风险;健康检查是部署成功的关键判断依据,建议在 CI 中等待健康检查通过后再标记部署完成;如果健康检查失败,应自动回滚到上一个稳定版本。

    8.2.11 部署代码完整示例(Dockerfile + docker-compose.yml)

    为了让部署过程更可复现,下面给出一个完整的部署代码示例,包含 Dockerfile、docker-compose.yml 和 .dockerignore,可直接用于生产环境:

    # Dockerfile
    # 使用 Python 3.11 作为基础镜像
    FROM python:3.11-slim

    # 设置工作目录
    WORKDIR /app

    # 设置环境变量
    ENV PYTHONDONTWRITEBYTECODE=1 \\
    PYTHONUNBUFFERED=1

    # 安装系统依赖(如需编译某些 Python 包)
    RUN apt-get update \\
    && apt-get install -y –no-install-recommends gcc \\
    && rm -rf /var/lib/apt/lists/*

    # 复制依赖文件并安装
    COPY requirements.txt .
    RUN pip install –no-cache-dir -r requirements.txt

    # 复制项目代码
    COPY . .

    # 暴露端口
    EXPOSE 8000

    # 启动命令
    CMD ["uvicorn", "main:app", "–host", "0.0.0.0", "–port", "8000"]

    # docker-compose.yml
    version: "3.8"

    services:
    api:
    build: .
    container_name: todoapi
    ports:
    "8000:8000"
    environment:
    DATABASE_URL=sqlite:///./todos.db
    SECRET_KEY=${SECRET_KEY}
    ACCESS_TOKEN_EXPIRE_MINUTES=30
    volumes:
    ./data:/app/data
    restart: unlessstopped
    healthcheck:
    test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
    interval: 30s
    timeout: 5s
    retries: 3
    start_period: 10s

    # .dockerignore
    __pycache__/
    *.pyc
    *.pyo
    *.db
    venv/
    .env
    .git/
    .pytest_cache/

    部署命令:

    # 构建并启动
    docker-compose up -d –build

    # 查看日志
    docker-compose logs -f api

    # 停止服务
    docker-compose down

    部署要点:使用 docker-compose 统一管理环境变量、端口映射和数据卷;healthcheck 让编排平台能自动检测服务健康状态;.dockerignore 避免将本地缓存和敏感文件打包进镜像;生产环境务必通过环境变量注入 SECRET_KEY 等敏感配置。

    8.2.12 部署步骤流程
    8.2.13 使用 Docker Compose 编排 API 与数据库(完整示例)

    前面 8.2.11 给出了单容器的 Dockerfile 和 docker-compose.yml。下面补充一个生产级多服务编排示例,将 API 服务与 PostgreSQL 数据库分离,并配置健康检查、数据卷和自动重启:

    # docker-compose.yml
    version: "3.8"

    services:
    api:
    build: .
    container_name: todoapi
    restart: unlessstopped
    ports:
    "8000:8000"
    environment:
    # 数据库连接指向 compose 网络中的 db 服务
    DATABASE_URL: postgresql://todo_user:todo_password@db:5432/todo_db
    SECRET_KEY: ${SECRET_KEY:pleasechangemeinproduction}
    ACCESS_TOKEN_EXPIRE_MINUTES: 30
    depends_on:
    db:
    condition: service_healthy
    healthcheck:
    test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
    interval: 30s
    timeout: 5s
    retries: 3
    start_period: 10s
    networks:
    todonetwork

    db:
    image: postgres:16alpine
    container_name: tododb
    restart: unlessstopped
    environment:
    POSTGRES_USER: todo_user
    POSTGRES_PASSWORD: todo_password
    POSTGRES_DB: todo_db
    volumes:
    postgres_data:/var/lib/postgresql/data
    healthcheck:
    test: ["CMD-SHELL", "pg_isready -U todo_user -d todo_db"]
    interval: 10s
    timeout: 5s
    retries: 5
    networks:
    todonetwork

    volumes:
    postgres_data:

    networks:
    todo-network:
    driver: bridge

    配套的 .env 文件(用于存放敏感配置,不要提交到 Git):

    # .env
    SECRET_KEY=your-super-secret-key-change-me
    POSTGRES_USER=todo_user
    POSTGRES_PASSWORD=your-strong-db-password

    配套的 Dockerfile(多阶段构建,减小镜像体积):

    # Dockerfile
    # 构建阶段
    FROM python:3.11-slim AS builder

    WORKDIR /app
    COPY requirements.txt .
    RUN pip install –no-cache-dir –prefix=/install -r requirements.txt

    # 运行阶段
    FROM python:3.11-slim

    WORKDIR /app
    COPY –from=builder /install /usr/local
    COPY . .

    # 非 root 用户运行,提升安全性
    RUN useradd -m appuser
    USER appuser

    EXPOSE 8000
    CMD ["uvicorn", "main:app", "–host", "0.0.0.0", "–port", "8000"]

    启动与验证:

    # 1. 构建并启动全部服务
    docker-compose up -d –build

    # 2. 查看服务状态
    docker-compose ps

    # 3. 查看 API 日志
    docker-compose logs -f api

    # 4. 验证健康检查
    curl http://localhost:8000/health

    # 5. 停止服务(保留数据卷)
    docker-compose down

    # 6. 彻底清理(删除数据卷)
    docker-compose down -v

    编排要点:depends_on 配合 condition: service_healthy 确保 API 在数据库就绪后才启动;数据卷 postgres_data 保证数据库重启不丢数据;restart: unless-stopped 让服务在异常退出后自动拉起;敏感配置通过 .env 注入,避免硬编码在代码中。

    下面用一张 Mermaid 流程图直观展示从本地开发到云平台上线的完整部署步骤,帮助你建立全局视角:

    #mermaid-svg-r7UEsSr7uY8b2lQ9{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-r7UEsSr7uY8b2lQ9 .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .error-icon{fill:#552222;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .marker{fill:#333333;stroke:#333333;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .marker.cross{stroke:#333333;}#mermaid-svg-r7UEsSr7uY8b2lQ9 svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-r7UEsSr7uY8b2lQ9 p{margin:0;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .cluster-label text{fill:#333;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .cluster-label span{color:#333;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .cluster-label span p{background-color:transparent;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .label text,#mermaid-svg-r7UEsSr7uY8b2lQ9 span{fill:#333;color:#333;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .node rect,#mermaid-svg-r7UEsSr7uY8b2lQ9 .node circle,#mermaid-svg-r7UEsSr7uY8b2lQ9 .node ellipse,#mermaid-svg-r7UEsSr7uY8b2lQ9 .node polygon,#mermaid-svg-r7UEsSr7uY8b2lQ9 .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .rough-node .label text,#mermaid-svg-r7UEsSr7uY8b2lQ9 .node .label text,#mermaid-svg-r7UEsSr7uY8b2lQ9 .image-shape .label,#mermaid-svg-r7UEsSr7uY8b2lQ9 .icon-shape .label{text-anchor:middle;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .rough-node .label,#mermaid-svg-r7UEsSr7uY8b2lQ9 .node .label,#mermaid-svg-r7UEsSr7uY8b2lQ9 .image-shape .label,#mermaid-svg-r7UEsSr7uY8b2lQ9 .icon-shape .label{text-align:center;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .node.clickable{cursor:pointer;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .arrowheadPath{fill:#333333;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-r7UEsSr7uY8b2lQ9 .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-r7UEsSr7uY8b2lQ9 .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-r7UEsSr7uY8b2lQ9 .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .cluster text{fill:#333;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .cluster span{color:#333;}#mermaid-svg-r7UEsSr7uY8b2lQ9 div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-r7UEsSr7uY8b2lQ9 rect.text{fill:none;stroke-width:0;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .icon-shape,#mermaid-svg-r7UEsSr7uY8b2lQ9 .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .icon-shape p,#mermaid-svg-r7UEsSr7uY8b2lQ9 .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .icon-shape .label rect,#mermaid-svg-r7UEsSr7uY8b2lQ9 .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-r7UEsSr7uY8b2lQ9 .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-r7UEsSr7uY8b2lQ9 .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-r7UEsSr7uY8b2lQ9 :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

    新手推荐

    灵活可控

    按量计费

    本地开发完成

    编写 Dockerfile

    本地构建镜像并验证

    选择部署方式

    Railway / Render 平台

    云服务器 + Docker

    Serverless 平台

    连接 GitHub 仓库

    平台自动构建部署

    配置环境变量

    获取公网 URL

    购买云服务器

    安装 Docker 环境

    上传镜像 / 拉取代码

    docker-compose up -d

    配置域名与 HTTPS

    编写 serverless 配置

    部署函数 / 容器

    配置触发器与限流

    部署后验证

    健康检查接口

    功能冒烟测试

    监控与日志

    上线完成

    部署流程要点:无论选择哪种方式,核心步骤都是「容器化 → 推送镜像/代码 → 平台构建 → 配置环境变量 → 获取公网地址 → 验证上线」;新手建议从 Railway / Render 开始,先跑通全流程,再逐步迁移到云服务器或 Serverless 方案。

    8.2.14 使用 Docker Compose 部署到云服务器(完整可运行
    8.2.15 部署验证脚本(完整可运行)

    在 8.2.14 的基础上,补充一个部署后的自动化验证脚本,用于确认云服务器上的 API 服务、数据库连接、健康检查、HTTPS 与反向代理均正常工作。

    #!/usr/bin/env bash
    # deploy_verify.sh
    # 部署验证脚本:检查 API 服务、数据库、健康检查、HTTPS、反向代理
    # 用法:./deploy_verify.sh [服务器IP或域名]

    set -e

    SERVER="${1:-your-server-ip}"
    BASE_URL="https://${SERVER}"
    echo "===== 部署验证开始:${BASE_URL} ====="

    # 1. 检查服务是否在线(HTTP 层)
    echo ""
    echo "[1/6] 检查服务在线状态…"
    if curl -sS -o /dev/null -w "%{http_code}" "${BASE_URL}/" | grep -q "200"; then
    echo " ✅ 服务在线,返回 200"
    else
    echo " ❌ 服务不可达,请检查 Docker 容器状态"
    exit 1
    fi

    # 2. 检查健康检查接口
    echo ""
    echo "[2/6] 检查健康检查接口…"
    HEALTH=$(curl -sS "${BASE_URL}/health")
    if echo "${HEALTH}" | grep -q '"status": "ok"'; then
    echo " ✅ 健康检查通过:${HEALTH}"
    else
    echo " ❌ 健康检查失败:${HEALTH}"
    exit 1
    fi

    # 3. 检查数据库连接(通过创建一条测试数据)
    echo ""
    echo "[3/6] 检查数据库读写…"
    CREATE_RESP=$(curl -sS -X POST "${BASE_URL}/todos" \\
    -H "Content-Type: application/json" \\
    -d '{"title": "部署验证-临时数据", "description": "验证后删除"}')

    TODO_ID=$(echo "${CREATE_RESP}" | python3 -c "import sys,json; print(json.load(sys.stdin)['id'])" 2>/dev/null || echo "")
    if [ -n "${TODO_ID}" ]; then
    echo " ✅ 数据库写入成功,ID=${TODO_ID}"
    # 清理测试数据
    curl -sS -X DELETE "${BASE_URL}/todos/${TODO_ID}" -o /dev/null
    echo " ✅ 测试数据已清理"
    else
    echo " ❌ 数据库写入失败:${CREATE_RESP}"
    exit 1
    fi

    # 4. 检查 HTTPS 证书
    echo ""
    echo "[4/6] 检查 HTTPS 证书…"
    CERT_EXPIRY=$(echo | openssl s_client -servername "${SERVER}" -connect "${SERVER}:443" 2>/dev/null \\
    | openssl x509 -noout -enddate 2>/dev/null | cut -d= -f2)

    if [ -n "${CERT_EXPIRY}" ]; then
    echo " ✅ 证书有效,到期时间:${CERT_EXPIRY}"
    else
    echo " ⚠️ 无法获取证书信息,请检查 Nginx HTTPS 配置"
    fi

    # 5. 检查反向代理(确认请求经过 Nginx)
    echo ""
    echo "[5/6] 检查反向代理…"
    PROXY_HEADER=$(curl -sSI "${BASE_URL}/" | grep -i "server" || true)
    if echo "${PROXY_HEADER}" | grep -qi "nginx"; then
    echo " ✅ 反向代理正常:${PROXY_HEADER}"
    else
    echo " ⚠️ 未检测到 Nginx 响应头:${PROXY_HEADER:-空}"
    fi

    # 6. 检查 Docker 容器状态
    echo ""
    echo "[6/6] 检查 Docker 容器状态…"
    if command -v docker &>/dev/null; then
    docker ps –format "table {{.Names}}\\t{{.Status}}\\t{{.Ports}}" | head -10
    else
    echo " ⚠️ 服务器上未安装 docker CLI,跳过容器检查"
    fi

    echo ""
    echo "===== 部署验证完成 ====="

    配套的 docker-compose.prod.yml 健康检查配置(在 8.2.14 基础上补充):

    version: "3.8"

    services:
    api:
    build: .
    container_name: todoapi
    restart: always
    environment:
    DATABASE_URL=sqlite:///./todos.db
    volumes:
    ./data:/app/data
    healthcheck:
    test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
    interval: 30s
    timeout: 5s
    retries: 3
    start_period: 10s
    networks:
    appnetwork

    nginx:
    image: nginx:1.25alpine
    container_name: todonginx
    restart: always
    ports:
    "80:80"
    "443:443"
    volumes:
    ./nginx.conf:/etc/nginx/nginx.conf:ro
    ./ssl:/etc/nginx/ssl:ro
    depends_on:
    api:
    condition: service_healthy
    networks:
    appnetwork

    networks:
    app-network:
    driver: bridge

    运行方式:

    # 在云服务器上执行
    chmod +x deploy_verify.sh
    ./deploy_verify.sh your-server-ip

    # 或指定域名
    ./deploy_verify.sh api.example.com

    验证要点:脚本按「服务在线 → 健康检查 → 数据库读写 → HTTPS 证书 → 反向代理 → Docker 容器」的顺序逐项验证,任何一步失败都会立即退出并给出明确提示;健康检查配置确保 Nginx 只在 API 就绪后才开始转发流量,避免启动期间的 502 错误。

    在 8.2.13 的基础上,补充一套可直接部署到云服务器(阿里云 / 腾讯云)的完整 Docker Compose 配置,包含 Nginx 反向代理、HTTPS 证书挂载、健康检查与自动重启策略。

    # docker-compose.prod.yml
    # 生产环境部署:API + 数据库 + Nginx 反向代理
    version: "3.8"

    services:
    api:
    build: .
    container_name: todoapi
    restart: unlessstopped
    environment:
    DATABASE_URL=postgresql://todo_user:todo_password@db:5432/todo_db
    SECRET_KEY=${SECRET_KEY}
    ACCESS_TOKEN_EXPIRE_MINUTES=30
    depends_on:
    db:
    condition: service_healthy
    expose:
    "8000"
    healthcheck:
    test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
    interval: 30s
    timeout: 5s
    retries: 3
    start_period: 10s
    networks:
    appnetwork

    db:
    image: postgres:16alpine
    container_name: tododb
    restart: unlessstopped
    environment:
    POSTGRES_USER=todo_user
    POSTGRES_PASSWORD=${DB_PASSWORD}
    POSTGRES_DB=todo_db
    volumes:
    postgres_data:/var/lib/postgresql/data
    healthcheck:
    test: ["CMD-SHELL", "pg_isready -U todo_user -d todo_db"]
    interval: 10s
    timeout: 5s
    retries: 5
    networks:
    appnetwork

    nginx:
    image: nginx:1.27alpine
    container_name: todonginx
    restart: unlessstopped
    ports:
    "80:80"
    "443:443"
    volumes:
    ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
    ./nginx/ssl:/etc/nginx/ssl:ro
    depends_on:
    api
    networks:
    appnetwork

    volumes:
    postgres_data:

    networks:
    app-network:
    driver: bridge

    # nginx/nginx.conf
    # Nginx 反向代理配置:HTTP 跳转 HTTPS + 转发到 API
    events {
    worker_connections 1024;
    }

    http {
    # HTTP 请求全部跳转到 HTTPS
    server {
    listen 80;
    server_name api.example.com;
    return 301 https://$host$request_uri;
    }

    # HTTPS 反向代理
    server {
    listen 443 ssl;
    server_name api.example.com;

    # SSL 证书(使用 certbot 生成后挂载)
    ssl_certificate /etc/nginx/ssl/fullchain.pem;
    ssl_certificate_key /etc/nginx/ssl/privkey.pem;
    ssl_protocols TLSv1.2 TLSv1.3;
    ssl_ciphers HIGH:!aNULL:!MD5;

    # 请求体大小限制(与 API 配置一致)
    client_max_body_size 10m;

    location / {
    proxy_pass http://api:8000;
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;
    }
    }
    }

    # 部署步骤
    # 1. 在云服务器上安装 Docker 和 Docker Compose
    curl -fsSL https://get.docker.com | sh
    sudo curl -L "https://github.com/docker/compose/releases/latest/download/docker-compose-$(uname -s)$(uname -m)" -o /usr/local/bin/docker-compose
    sudo chmod +x /usr/local/bin/docker-compose

    # 2. 克隆项目并创建环境变量文件
    git clone https://github.com/your-username/my-api.git
    cd my-api
    cp .env.example .env
    # 编辑 .env,填入 SECRET_KEY 和 DB_PASSWORD

    # 3. 生成 SSL 证书(使用 certbot)
    sudo apt install certbot
    sudo certbot certonly –standalone -d api.example.com
    # 将证书复制到 nginx/ssl 目录
    sudo cp /etc/letsencrypt/live/api.example.com/fullchain.pem nginx/ssl/
    sudo cp /etc/letsencrypt/live/api.example.com/privkey.pem nginx/ssl/

    # 4. 启动服务
    docker-compose -f docker-compose.prod.yml up -d –build

    # 5. 查看运行状态
    docker-compose -f docker-compose.prod.yml ps
    docker-compose -f docker-compose.prod.yml logs -f api

    # 6. 验证部署
    curl https://api.example.com/health

    # .env 环境变量示例
    SECRET_KEY=your-random-secret-key-here
    DB_PASSWORD=your-strong-db-password

    生产部署要点:API 容器不直接暴露端口,由 Nginx 统一对外提供 HTTPS 服务;PostgreSQL 数据通过命名卷持久化,容器重建不丢失;restart: unless-stopped 保证服务异常退出后自动重启;健康检查确保 API 就绪后才开始接收流量;SSL 证书使用 certbot 免费签发,并配置自动续期(certbot renew –dry-run 验证)。

    8. 部署上线

    API 开发完成后,下一步就是把它部署到公网,让其他用户能够访问。本章介绍两种主流的部署方式:Docker 容器化部署和云平台部署。

    8.1 使用 Docker 部署

    Docker 可以将应用及其依赖打包成镜像,实现「一次构建,到处运行」。首先在项目根目录创建 Dockerfile:

    # 使用 Python 3.11 作为基础镜像
    FROM python:3.11-slim

    # 设置工作目录
    WORKDIR /app

    # 复制依赖文件并安装
    COPY requirements.txt .
    RUN pip install –no-cache-dir -r requirements.txt

    # 复制项目代码
    COPY . .

    # 暴露端口
    EXPOSE 8000

    # 启动命令
    CMD ["uvicorn", "main:app", "–host", "0.0.0.0", "–port", "8000"]

    创建 requirements.txt 文件,列出所有依赖:

    fastapi==0.115.0
    uvicorn[standard]==0.30.6
    sqlalchemy==2.0.35
    pydantic==2.9.2
    python-jose[cryptography]==3.3.0
    passlib[bcrypt]==1.7.4
    python-multipart==0.0.9
    prometheus-fastapi-instrumentator==7.0.0

    然后构建并运行镜像:

    # 构建镜像
    docker build -t todo-api .

    # 运行容器
    docker run -d –name todo-api -p 8000:8000 todo-api

    启动后访问 http://localhost:8000 即可看到 API 正常运行。使用 docker logs -f todo-api 查看日志,docker stop todo-api 停止容器。

    8.2 部署到云平台

    除了自建服务器,还可以使用云平台快速部署,省去服务器运维的麻烦。

    8.2.1 使用 Railway / Render 部署(推荐新手)

    Railway 和 Render 是流行的 PaaS 平台,支持直接从 GitHub 仓库部署,免费额度足够个人项目使用。

    以 Render 为例:

  • 将代码推送到 GitHub 仓库
  • 在 Render 控制台点击「New +」→「Web Service」
  • 连接你的 GitHub 仓库
  • 配置启动命令:uvicorn main:app –host 0.0.0.0 –port $PORT
  • 点击「Create Web Service」即可完成部署
  • 部署完成后,Render 会分配一个公网 URL(如 https://todo-api.onrender.com),你的 API 就上线了。

    8.2.2 使用云服务器 + Docker 部署(阿里云 / 腾讯云)

    如果你已经有一台云服务器(如阿里云 ECS、腾讯云 CVM),可以按以下步骤部署:

    # 1. 在服务器上安装 Docker
    curl -fsSL https://get.docker.com | bash

    # 2. 将项目代码上传到服务器(使用 git 或 scp)
    git clone https://github.com/yourname/todo-api.git
    cd todo-api

    # 3. 构建并运行容器
    docker build -t todo-api .
    docker run -d –name todo-api -p 8000:8000 –restart=always todo-api

    注意:–restart=always 让容器在服务器重启后自动启动。记得在云控制台的安全组中放行 8000 端口。

    8.2.3 使用 Serverless 平台部署(按调用量计费)

    Serverless 平台(如阿里云函数计算、腾讯云 SCF)按调用次数计费,适合流量波动大的场景。以阿里云函数计算为例:

  • 在函数计算控制台创建函数,选择 Python 3.11 运行时
  • 将 main.py 和依赖打包上传
  • 配置函数入口为 main.app
  • 创建自定义域名,绑定到函数
  • Serverless 平台会自动扩缩容,无需关心服务器资源,但需要注意冷启动延迟和平台限制。

    8.2.4 部署注意事项
    • 环境变量:数据库连接串、JWT 密钥等敏感信息通过环境变量注入,不要硬编码在代码里
    • 端口配置:云平台通常会注入 PORT 环境变量,启动命令要使用 $PORT
    • 数据库迁移:生产环境使用 PostgreSQL/MySQL,部署前执行数据库迁移脚本
    • HTTPS:通过云平台或 Nginx 配置 HTTPS 证书,加密传输数据
    8.2.5 部署上线检查清单
    • 代码已推送到 Git 仓库
    • 依赖已锁定版本(requirements.txt)
    • 敏感信息已改为环境变量
    • 数据库迁移脚本已执行
    • 健康检查接口 /health 可访问
    • HTTPS 证书已配置
    • 日志和监控已接入
    8.2.6 部署上线完整流程示例

    下面给出一个完整的部署流程示例,从本地构建到云服务器上线:

    # 1. 本地构建并测试镜像
    docker build -t todo-api .
    docker run -d –name todo-api-test -p 8000:8000 todo-api
    curl http://localhost:8000/health

    # 2. 将镜像推送到镜像仓库(以 Docker Hub 为例)
    docker tag todo-api yourname/todo-api:latest
    docker push yourname/todo-api:latest

    # 3. 登录云服务器,拉取并运行镜像
    ssh root@your-server-ip
    docker pull yourname/todo-api:latest
    docker run -d –name todo-api \\
    -p 8000:8000 \\
    –restart=always \\
    -e DATABASE_URL="postgresql://user:pass@host:5432/todos" \\
    -e SECRET_KEY="your-secret-key" \\
    yourname/todo-api:latest

    # 4. 验证部署结果
    curl http://your-server-ip:8000/health

    部署成功标志:/health 返回 {"status": "ok"},且通过公网 IP 能正常访问 /docs 文档页面。

    8.2.7 部署后的验证与回滚策略

    服务上线后,建议按以下顺序进行验证,确保新版本稳定运行:

  • 健康检查:访问 /health 接口,确认返回 {"status": "ok"}
  • 功能冒烟测试:对核心 CRUD 接口各执行一次请求,确认读写正常
  • 监控指标检查:查看 Prometheus 或云平台监控面板,确认请求量、错误率、耗时在正常范围
  • 灰度观察:先放量 10% 流量观察 30 分钟,无异常再逐步放量到 100%
  • 回滚策略:

    • Docker 部署:保留上一版本镜像,通过 docker rollback 或重新部署旧镜像快速回滚
    • 云平台部署:Railway / Render 支持一键回滚到历史版本,在部署记录中选择即可
    • 数据库兼容:回滚前确认数据库表结构变更是否向后兼容,必要时先执行数据迁移脚本

    上线检查口诀:先看健康、再测功能、后看监控、留好回滚。任何一步异常都立即回滚,不要带病上线。

    8.2.8 使用 Docker Compose 编排多服务

    当 API 需要依赖数据库、缓存等外部服务时,使用 Docker Compose 可以一键启动整个环境。下面演示如何编排 FastAPI + PostgreSQL 的组合。

    创建 docker-compose.yml 文件:

    version: "3.8"

    services:
    api:
    build: .
    container_name: todoapi
    ports:
    "8000:8000"
    environment:
    DATABASE_URL=postgresql://todo:todo123@db:5432/todos
    SECRET_KEY=${SECRET_KEY}
    depends_on:
    db
    restart: unlessstopped

    db:
    image: postgres:15
    container_name: tododb
    environment:
    POSTGRES_USER=todo
    POSTGRES_PASSWORD=todo123
    POSTGRES_DB=todos
    volumes:
    pgdata:/var/lib/postgresql/data
    restart: unlessstopped

    volumes:
    pgdata:

    对应的 Dockerfile:

    FROM python:3.11-slim

    WORKDIR /app

    COPY requirements.txt .
    RUN pip install –no-cache-dir -r requirements.txt

    COPY . .

    EXPOSE 8000

    CMD ["uvicorn", "main:app", "–host", "0.0.0.0", "–port", "8000"]

    启动服务:

    # 构建并启动
    docker-compose up -d

    # 查看日志
    docker-compose logs -f api

    # 停止服务
    docker-compose down

    # 停止并删除数据卷(慎用,会清空数据库)
    docker-compose down -v

    Compose 要点:depends_on 确保数据库先启动;通过环境变量注入数据库连接串和密钥,避免硬编码;数据卷 pgdata 保证数据库重启后数据不丢失。

    8.2.9 部署代码完整示例

    下面提供一个从 Dockerfile 到 CI/CD 的完整部署代码示例,方便你直接复制使用。

    Dockerfile

    # 使用 Python 3.11 作为基础镜像
    FROM python:3.11-slim

    # 设置工作目录
    WORKDIR /app

    # 设置环境变量
    ENV PYTHONDONTWRITEBYTECODE=1 \\
    PYTHONUNBUFFERED=1

    # 安装系统依赖(如需编译某些 Python 包)
    RUN apt-get update \\
    && apt-get install -y –no-install-recommends gcc \\
    && rm -rf /var/lib/apt/lists/*

    # 复制依赖文件并安装
    COPY requirements.txt .
    RUN pip install –no-cache-dir -r requirements.txt

    # 复制项目代码
    COPY . .

    # 创建非 root 用户运行应用(安全最佳实践)
    RUN useradd -m appuser && chown -R appuser:appuser /app
    USER appuser

    # 暴露端口
    EXPOSE 8000

    # 启动命令
    CMD ["uvicorn", "main:app", "–host", "0.0.0.0", "–port", "8000"]

    docker-compose.yml

    version: "3.8"

    services:
    api:
    build: .
    container_name: myapi
    ports:
    "8000:8000"
    environment:
    DATABASE_URL=sqlite:///./todos.db
    SECRET_KEY=${SECRET_KEY}
    volumes:
    ./data:/app/data
    restart: unlessstopped
    healthcheck:
    test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
    interval: 30s
    timeout: 5s
    retries: 3

    CI/CD 流水线(GitHub Actions)

    # .github/workflows/deploy.yml
    name: Deploy API

    on:
    push:
    branches: [main]

    jobs:
    test:
    runs-on: ubuntulatest
    steps:
    uses: actions/checkout@v4
    uses: actions/setuppython@v5
    with:
    python-version: "3.11"
    run: pip install r requirements.txt
    run: pytest tests/ v

    deploy:
    needs: test
    runs-on: ubuntulatest
    steps:
    uses: actions/checkout@v4
    name: Deploy to Railway
    uses: railwayapp/railwayaction@v2
    with:
    railway_token: ${{ secrets.RAILWAY_TOKEN }}
    service: myapi

    部署代码要点:Dockerfile 使用非 root 用户运行提升安全性;docker-compose 通过环境变量注入敏感配置;CI/CD 流水线先跑测试再部署,确保只有通过测试的代码才会发布。 现

    8.3 部署后的监控

    API 上线后,监控是保障服务稳定运行的关键一环。下面介绍几个实用的监控手段。

    8.3.1 健康检查接口

    在 main.py 中添加一个健康检查接口,方便负载均衡器和监控系统定期探测服务状态:

    @app.get("/health")
    def health_check():
    """健康检查接口,返回服务运行状态"""
    return {"status": "ok", "service": "todo-api", "version": "1.0.0"}

    8.3.2 结构化日志

    使用 Python 内置的 logging 模块记录请求日志,方便排查问题:

    import logging
    import time
    from fastapi import Request

    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger("todo-api")

    @app.middleware("http")
    async def log_requests(request: Request, call_next):
    """记录每个请求的方法、路径、状态码和耗时"""
    start = time.time()
    response = await call_next(request)
    duration = time.time() start
    logger.info(
    f"{request.method} {request.url.path} -> {response.status_code} "
    f"({duration:.3f}s)"
    )
    return response

    8.3.3 接入监控平台
    • Prometheus + Grafana:通过 prometheus-fastapi-instrumentator 暴露 /metrics 指标,用 Grafana 可视化 CPU、内存、请求量、错误率等
    • Sentry:接入后自动捕获未处理异常并上报,配合告警规则第一时间收到通知
    • Uptime Robot / 阿里云云监控:定时探测 /health 接口,服务不可用时立即告警
    8.3.4 告警与日志收集
    • 将容器日志输出到 stdout,由 Docker 或云平台统一收集
    • 使用 ELK(Elasticsearch + Logstash + Kibana)或 Loki 集中存储和检索日志
    • 设置告警规则:如 5 分钟内的 5xx 错误率超过 1% 时触发通知
    8.3.5 监控指标采集

    除了日志,还需要采集关键运行指标,用于容量规划和性能调优:

    # 安装依赖
    # pip install prometheus-fastapi-instrumentator

    from prometheus_fastapi_instrumentator import Instrumentator

    # 在创建 app 后立即初始化
    Instrumentator().instrument(app).expose(app)

    启动后访问 http://127.0.0.1:8000/metrics 即可看到 Prometheus 格式的指标,包括:

    • 请求量:http_requests_total,按方法、路径、状态码维度统计
    • 请求耗时:http_request_duration_seconds,含平均值、分位数(P50/P95/P99)
    • 进程指标:CPU 使用率、内存占用、文件描述符数量

    监控指标建议:重点关注 P95/P99 耗时和 5xx 错误率。P99 超过 500ms 或错误率持续高于 1% 时,应触发告警并排查。

    8.3.6 日志采集与集中管理

    当服务实例增多后,分散在每台机器上的日志难以检索,需要集中采集。下面以 Filebeat 和 Loki 为例,给出完整的配置示例。

    方案一:Filebeat + Elasticsearch + Kibana(ELK)

    Filebeat 负责采集日志文件并发送到 Elasticsearch,Kibana 负责可视化检索。首先安装并配置 Filebeat:

    # filebeat.yml
    filebeat.inputs:
    type: filestream
    id: todoapilogs
    enabled: true
    paths:
    /var/log/todoapi/*.log
    fields:
    service: todoapi
    env: production

    output.elasticsearch:
    hosts: ["https://elasticsearch:9200"]
    username: "${ES_USERNAME}"
    password: "${ES_PASSWORD}"

    setup.kibana:
    host: "https://kibana:5601"

    启动 Filebeat 后,日志会自动流入 Elasticsearch,你可以在 Kibana 中按关键字、时间范围、服务实例等维度检索日志。

    方案二:Promtail + Loki + Grafana

    Loki 是 Grafana 生态的轻量级日志聚合系统,与 Prometheus 共用 Grafana 面板。在 promtail.yml 中配置日志采集:

    # promtail.yml
    server:
    http_listen_port: 9080
    grpc_listen_port: 0

    positions:
    filename: /tmp/positions.yaml

    clients:
    url: http://loki:3100/loki/api/v1/push

    scrape_configs:
    job_name: todoapi
    static_configs:
    targets:
    localhost
    labels:
    job: todoapi
    __path__: /var/log/todoapi/*.log

    采集后即可在 Grafana 中按关键字、时间范围、服务实例等维度检索日志,快速定位问题。

    日志规范建议

    无论使用哪种方案,都建议遵循以下规范:

    • 统一格式:使用 JSON 结构化日志,便于字段检索和过滤
    • 包含关键字段:时间戳、日志级别、服务名、请求 ID、用户 ID、耗时等
    • 分级存储:热日志保留 7 天,冷日志归档到对象存储,控制存储成本
    • 敏感信息脱敏:日志中不要记录密码、token、手机号等敏感信息,必要时做脱敏处理
    8.3.7 告警规则示例

    在 Prometheus 中配置告警规则,当指标异常时自动通知:

    groups:
    name: todoapialerts
    rules:
    alert: HighErrorRate
    expr: |
    sum(rate(http_requests_total{status=~"5.."}[5m]))
    / sum(rate(http_requests_total[5m])) > 0.01

    for: 5m
    labels:
    severity: critical
    annotations:
    summary: "5xx 错误率超过 1%"
    alert: HighLatency
    expr: |
    histogram_quantile(0.99, rate(http_request_duration_seconds_bucket[5m])) > 0.5

    for: 5m
    labels:
    severity: warning
    annotations:
    summary: "P99 耗时超过 500ms"

    告警分级:建议将告警分为 critical(立即处理)、warning(工作时间处理)、info(记录观察)三级,避免告警疲劳。

    8.3.8 部署监控代码示例

    下面给出一个完整的监控代码示例,包含健康检查、指标采集和日志收集。

    健康检查接口

    # main.py 中添加健康检查接口
    from fastapi import FastAPI
    from sqlalchemy import text

    app = FastAPI(title="待办事项 API")

    @app.get("/health")
    def health_check():
    """健康检查接口,供负载均衡器和监控平台调用"""
    try:
    # 检查数据库连接
    db.execute(text("SELECT 1"))
    db_status = "ok"
    except Exception:
    db_status = "error"

    return {
    "status": "ok" if db_status == "ok" else "degraded",
    "database": db_status,
    "version": app.version,
    }

    Prometheus 指标采集

    # metrics.py
    """
    Prometheus 指标采集
    统计请求数、耗时、错误率等关键指标
    """

    from prometheus_client import Counter, Histogram, generate_latest
    from fastapi import Request
    from fastapi.responses import Response
    import time

    # 定义指标
    REQUEST_COUNT = Counter(
    "http_requests_total",
    "Total HTTP requests",
    ["method", "endpoint", "status"],
    )

    REQUEST_DURATION = Histogram(
    "http_request_duration_seconds",
    "HTTP request duration in seconds",
    ["method", "endpoint"],
    )

    @app.middleware("http")
    async def metrics_middleware(request: Request, call_next):
    """记录每个请求的指标"""
    start_time = time.time()
    response = await call_next(request)
    duration = time.time() start_time

    REQUEST_COUNT.labels(
    method=request.method,
    endpoint=request.url.path,
    status=response.status_code,
    ).inc()

    REQUEST_DURATION.labels(
    method=request.method,
    endpoint=request.url.path,
    ).observe(duration)

    return response

    @app.get("/metrics")
    def metrics():
    """Prometheus 抓取指标端点"""
    return Response(generate_latest(), media_type="text/plain")

    结构化日志配置

    # logging_config.py
    """
    结构化日志配置
    输出 JSON 格式日志,便于日志平台采集和分析
    """

    import json
    import logging
    from datetime import datetime

    class JSONFormatter(logging.Formatter):
    """将日志格式化为 JSON"""

    def format(self, record):
    log_entry = {
    "timestamp": datetime.utcnow().isoformat(),
    "level": record.levelname,
    "logger": record.name,
    "message": record.getMessage(),
    }
    # 添加额外字段
    if hasattr(record, "request_id"):
    log_entry["request_id"] = record.request_id
    if hasattr(record, "user_id"):
    log_entry["user_id"] = record.user_id
    if record.exc_info:
    log_entry["exception"] = self.formatException(record.exc_info)
    return json.dumps(log_entry, ensure_ascii=False)

    def setup_logging():
    """配置全局日志"""
    handler = logging.StreamHandler()
    handler.setFormatter(JSONFormatter())
    root_logger = logging.getLogger()
    root_logger.addHandler(handler)
    root_logger.setLevel(logging.INFO)

    监控代码要点:健康检查接口返回数据库连接状态;Prometheus 指标覆盖请求量、耗时和错误率;结构化日志输出 JSON 格式,方便接入 ELK 或 Loki 等日志平台。

    8.3.9 部署监控架构图

    下面用一张 Mermaid 架构图,直观展示从客户端请求到日志、指标采集,再到告警通知的完整监控数据流:

    #mermaid-svg-qUpAgdZeryXbNiFv{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-qUpAgdZeryXbNiFv .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-qUpAgdZeryXbNiFv .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-qUpAgdZeryXbNiFv .error-icon{fill:#552222;}#mermaid-svg-qUpAgdZeryXbNiFv .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-qUpAgdZeryXbNiFv .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-qUpAgdZeryXbNiFv .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-qUpAgdZeryXbNiFv .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-qUpAgdZeryXbNiFv .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-qUpAgdZeryXbNiFv .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-qUpAgdZeryXbNiFv .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-qUpAgdZeryXbNiFv .marker{fill:#333333;stroke:#333333;}#mermaid-svg-qUpAgdZeryXbNiFv .marker.cross{stroke:#333333;}#mermaid-svg-qUpAgdZeryXbNiFv svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-qUpAgdZeryXbNiFv p{margin:0;}#mermaid-svg-qUpAgdZeryXbNiFv .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-qUpAgdZeryXbNiFv .cluster-label text{fill:#333;}#mermaid-svg-qUpAgdZeryXbNiFv .cluster-label span{color:#333;}#mermaid-svg-qUpAgdZeryXbNiFv .cluster-label span p{background-color:transparent;}#mermaid-svg-qUpAgdZeryXbNiFv .label text,#mermaid-svg-qUpAgdZeryXbNiFv span{fill:#333;color:#333;}#mermaid-svg-qUpAgdZeryXbNiFv .node rect,#mermaid-svg-qUpAgdZeryXbNiFv .node circle,#mermaid-svg-qUpAgdZeryXbNiFv .node ellipse,#mermaid-svg-qUpAgdZeryXbNiFv .node polygon,#mermaid-svg-qUpAgdZeryXbNiFv .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-qUpAgdZeryXbNiFv .rough-node .label text,#mermaid-svg-qUpAgdZeryXbNiFv .node .label text,#mermaid-svg-qUpAgdZeryXbNiFv .image-shape .label,#mermaid-svg-qUpAgdZeryXbNiFv .icon-shape .label{text-anchor:middle;}#mermaid-svg-qUpAgdZeryXbNiFv .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-qUpAgdZeryXbNiFv .rough-node .label,#mermaid-svg-qUpAgdZeryXbNiFv .node .label,#mermaid-svg-qUpAgdZeryXbNiFv .image-shape .label,#mermaid-svg-qUpAgdZeryXbNiFv .icon-shape .label{text-align:center;}#mermaid-svg-qUpAgdZeryXbNiFv .node.clickable{cursor:pointer;}#mermaid-svg-qUpAgdZeryXbNiFv .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-qUpAgdZeryXbNiFv .arrowheadPath{fill:#333333;}#mermaid-svg-qUpAgdZeryXbNiFv .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-qUpAgdZeryXbNiFv .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-qUpAgdZeryXbNiFv .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-qUpAgdZeryXbNiFv .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-qUpAgdZeryXbNiFv .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-qUpAgdZeryXbNiFv .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-qUpAgdZeryXbNiFv .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-qUpAgdZeryXbNiFv .cluster text{fill:#333;}#mermaid-svg-qUpAgdZeryXbNiFv .cluster span{color:#333;}#mermaid-svg-qUpAgdZeryXbNiFv div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-qUpAgdZeryXbNiFv .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-qUpAgdZeryXbNiFv rect.text{fill:none;stroke-width:0;}#mermaid-svg-qUpAgdZeryXbNiFv .icon-shape,#mermaid-svg-qUpAgdZeryXbNiFv .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-qUpAgdZeryXbNiFv .icon-shape p,#mermaid-svg-qUpAgdZeryXbNiFv .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-qUpAgdZeryXbNiFv .icon-shape .label rect,#mermaid-svg-qUpAgdZeryXbNiFv .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-qUpAgdZeryXbNiFv .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-qUpAgdZeryXbNiFv .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-qUpAgdZeryXbNiFv :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

    通知层

    监控与存储层

    应用服务层

    客户端

    HTTP 请求

    存活/就绪探针

    抓取指标

    采集日志

    用户 / 调用方

    FastAPI 应用

    健康检查接口 /health

    结构化日志(JSON)

    监控指标(Prometheus)

    Prometheus 指标库

    Grafana 可视化看板

    Loki 日志中心

    Alertmanager 告警

    邮件告警

    Webhook / 钉钉 / 企业微信

    架构说明:应用层通过 /health 暴露存活探针、输出结构化日志并暴露 Prometheus 指标;监控层负责指标抓取、日志采集与告警规则判定;通知层通过邮件或 Webhook 将异常及时推送给运维人员,形成「采集 → 存储 → 可视化 → 告警」的闭环。

    9. API 安全

    API 一旦上线公网,就会面临各种安全威胁。本章介绍 API 安全的核心实践,包括身份认证、授权、输入校验、限流等。

    9.1 身份认证

    身份认证解决「你是谁」的问题。目前最常用的方案是 JWT(JSON Web Token),它无状态、可扩展,非常适合分布式系统。

    9.1.1 安装依赖

    pip install pyjwt passlib[bcrypt] python-multipart

    9.1.2 用户模型与密码哈希

    from passlib.context import CryptContext

    # 使用 bcrypt 哈希密码
    pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

    def hash_password(password: str) > str:
    """对明文密码进行哈希"""
    return pwd_context.hash(password)

    def verify_password(plain_password: str, hashed_password: str) > bool:
    """校验密码是否正确"""
    return pwd_context.verify(plain_password, hashed_password)

    9.1.3 JWT 工具函数

    import jwt
    from datetime import datetime, timedelta
    from fastapi import Header, HTTPException

    SECRET_KEY = "your-secret-key"
    ALGORITHM = "HS256"
    TOKEN_EXPIRE_MINUTES = 60

    def create_token(user_id: int) > str:
    """生成 JWT token"""
    payload = {
    "sub": str(user_id),
    "exp": datetime.utcnow() + timedelta(minutes=TOKEN_EXPIRE_MINUTES),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

    def verify_token(authorization: str = Header(...)):
    """校验 token,返回当前用户 ID"""
    if not authorization.startswith("Bearer "):
    raise HTTPException(status_code=401, detail="无效的认证头")
    token = authorization.split(" ")[1]
    try:
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    return int(payload["sub"])
    except jwt.ExpiredSignatureError:
    raise HTTPException(status_code=401, detail="token 已过期")
    except jwt.InvalidTokenError:
    raise HTTPException(status_code=401, detail="无效的 token")

    9.1.4 注册与登录接口

    from pydantic import BaseModel

    class UserCreate(BaseModel):
    """注册请求体"""
    username: str
    password: str

    class UserLogin(BaseModel):
    """登录请求体"""
    username: str
    password: str

    # 模拟用户表,实际项目中应存入数据库并哈希密码
    fake_users_db = {}

    @app.post("/register", status_code=201)
    def register(user: UserCreate):
    """用户注册,返回新用户 ID"""
    if user.username in fake_users_db:
    raise HTTPException(status_code=400, detail="用户名已存在")
    user_id = len(fake_users_db) + 1
    # 生产环境务必使用 bcrypt 等慢哈希算法存储密码,切勿明文保存
    fake_users_db[user.username] = {"id": user_id, "password": user.password}
    return {"id": user_id, "username": user.username}

    @app.post("/login")
    def login(user: UserLogin):
    """用户登录,校验通过后返回 JWT token"""
    record = fake_users_db.get(user.username)
    if not record or record["password"] != user.password:
    raise HTTPException(status_code=401, detail="用户名或密码错误")
    token = create_token(record["id"])
    return {"access_token": token, "token_type": "bearer"}

    9.1.5 保护受保护接口

    @app.get("/me")
    def get_me(user_id: int = Depends(verify_token)):
    """获取当前登录用户信息(需携带 token)"""
    for username, record in fake_users_db.items():
    if record["id"] == user_id:
    return {"id": user_id, "username": username}
    raise HTTPException(status_code=404, detail="用户不存在")

    调用方式:

    # 1. 注册
    curl -X POST "http://127.0.0.1:8000/register" \\
    -H "Content-Type: application/json" \\
    -d '{"username": "alice", "password": "pass1234"}'

    # 2. 登录,拿到 token
    curl -X POST "http://127.0.0.1:8000/login" \\
    -H "Content-Type: application/json" \\
    -d '{"username": "alice", "password": "pass1234"}'

    # 3. 携带 token 访问受保护接口
    curl "http://127.0.0.1:8000/me" \\
    -H "Authorization: Bearer <上一步返回的token>"

    认证闭环要点:注册时校验用户名唯一性,登录时校验密码并签发 token,受保护接口通过 Depends(verify_token) 统一鉴权。生产环境请务必使用 bcrypt 哈希密码、将用户表落到数据库,并配合限流策略防暴力破解。

    9. API 安全

    API 上线后,安全是必须优先考虑的问题。一个没有安全防护的 API 就像敞开的大门,任何人都可以随意访问和操作数据。本章将从身份认证、授权、输入校验、限流、日志审计等方面,系统讲解如何为你的 API 加上安全防护。

    9.1 身份认证

    身份认证解决「你是谁」的问题。目前最常用的认证方式是 JWT(JSON Web Token),它无状态、可扩展,非常适合 RESTful API。

    9.1.1 安装依赖

    pip install python-jose[cryptography] passlib[bcrypt] python-multipart

    • python-jose:用于生成和校验 JWT
    • passlib[bcrypt]:用于密码哈希存储
    • python-multipart:用于处理 OAuth2 表单登录
    9.1.2 用户模型与密码哈希

    在 main.py 中新增用户模型和密码工具:

    from passlib.context import CryptContext
    from pydantic import BaseModel

    # 密码哈希工具
    pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

    class UserCreate(BaseModel):
    """注册请求体"""
    username: str
    password: str

    class UserLogin(BaseModel):
    """登录请求体"""
    username: str
    password: str

    def hash_password(password: str) > str:
    """对明文密码进行 bcrypt 哈希"""
    return pwd_context.hash(password)

    def verify_password(plain_password: str, hashed_password: str) > bool:
    """校验明文密码与哈希是否匹配"""
    return pwd_context.verify(plain_password, hashed_password)

    为什么用 bcrypt:明文存储密码是极其危险的做法,一旦数据库泄露,所有用户密码都会暴露。bcrypt 是慢哈希算法,能显著增加暴力破解的成本。

    9.1.3 JWT 工具函数

    from datetime import datetime, timedelta
    from jose import jwt, JWTError
    from fastapi import Header, HTTPException

    # 生产环境务必通过环境变量注入,不要硬编码
    SECRET_KEY = "your-secret-key"
    ALGORITHM = "HS256"
    TOKEN_EXPIRE_MINUTES = 60

    def create_token(user_id: int) > str:
    """生成 JWT token"""
    payload = {
    "sub": str(user_id),
    "exp": datetime.utcnow() + timedelta(minutes=TOKEN_EXPIRE_MINUTES),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

    def verify_token(authorization: str = Header(...)):
    """校验 token,返回当前用户 ID"""
    if not authorization.startswith("Bearer "):
    raise HTTPException(status_code=401, detail="无效的认证头")
    token = authorization.split(" ")[1]
    try:
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    return int(payload["sub"])
    except jwt.ExpiredSignatureError:
    raise HTTPException(status_code=401, detail="token 已过期")
    except jwt.InvalidTokenError:
    raise HTTPException(status_code=401, detail="无效的 token")

    9.1.4 注册与登录接口

    # 模拟用户表,实际项目中应存入数据库
    fake_users_db = {}

    @app.post("/register", status_code=201)
    def register(user: UserCreate):
    """用户注册,返回新用户 ID"""
    if user.username in fake_users_db:
    raise HTTPException(status_code=400, detail="用户名已存在")
    user_id = len(fake_users_db) + 1
    fake_users_db[user.username] = {
    "id": user_id,
    "username": user.username,
    "password": hash_password(user.password), # 存储哈希,不存明文
    }
    return {"id": user_id, "username": user.username}

    @app.post("/login")
    def login(user: UserLogin):
    """用户登录,校验通过后返回 JWT token"""
    record = fake_users_db.get(user.username)
    if not record or not verify_password(user.password, record["password"]):
    raise HTTPException(status_code=401, detail="用户名或密码错误")
    token = create_token(record["id"])
    return {"access_token": token, "token_type": "bearer"}

    9.1.5 保护受保护接口

    在需要登录才能访问的接口上,通过 Depends(verify_token) 注入当前用户:

    @app.get("/me")
    def get_me(user_id: int = Depends(verify_token)):
    """获取当前登录用户信息(需携带 token)"""
    for username, record in fake_users_db.items():
    if record["id"] == user_id:
    return {"id": user_id, "username": username}
    raise HTTPException(status_code=404, detail="用户不存在")

    调用方式:

    # 1. 注册
    curl -X POST "http://127.0.0.1:8000/register" \\
    -H "Content-Type: application/json" \\
    -d '{"username": "alice", "password": "pass1234"}'

    # 2. 登录,拿到 token
    curl -X POST "http://127.0.0.1:8000/login" \\
    -H "Content-Type: application/json" \\
    -d '{"username": "alice", "password": "pass1234"}'

    # 3. 携带 token 访问受保护接口
    curl "http://127.0.0.1:8000/me" \\
    -H "Authorization: Bearer <上一步返回的token>"

    认证闭环要点:注册时对密码做 bcrypt 哈希,登录时校验密码并签发 JWT,受保护接口通过 Depends(verify_token) 统一鉴权。生产环境请务必将用户表落到数据库,并配合 9.4 节的限流策略防暴力破解。

    9.2 授权(RBAC)

    认证解决「你是谁」,授权解决「你能做什么」。RBAC(基于角色的访问控制)是最常见的授权模型。下面演示如何为接口添加角色权限控制。

    from enum import Enum
    from fastapi import Depends, HTTPException

    class UserRole(str, Enum):
    """用户角色"""
    ADMIN = "admin"
    USER = "user"

    # 在用户表中增加角色字段
    fake_users_db = {
    "alice": {"id": 1, "username": "alice", "password": "…", "role": UserRole.ADMIN},
    "bob": {"id": 2, "username": "bob", "password": "…", "role": UserRole.USER},
    }

    def require_role(required_role: UserRole):
    """角色校验依赖工厂"""
    def role_checker(user_id: int = Depends(verify_token)) > int:
    for record in fake_users_db.values():
    if record["id"] == user_id:
    if record["role"] != required_role:
    raise HTTPException(status_code=403, detail="权限不足")
    return user_id
    raise HTTPException(status_code=404, detail="用户不存在")
    return role_checker

    # 只有管理员可以删除待办事项
    @app.delete("/todos/{todo_id}", status_code=204)
    def delete_todo_admin(todo_id: int, user_id: int = Depends(require_role(UserRole.ADMIN))):
    """删除待办事项(仅管理员)"""
    # 删除逻辑…
    pass

    授权设计建议:将角色定义在数据库表中,通过中间表关联用户与角色,便于灵活扩展。接口层通过 Depends(require_role(…)) 声明所需角色,实现声明式权限控制。

    9.3 输入校验与安全防护

    除了认证和授权,还需要对输入数据做严格校验,防止注入攻击和恶意请求。

    9.3.1 使用 Pydantic 严格校验

    from pydantic import BaseModel, Field, validator

    class TodoCreate(BaseModel):
    """创建待办事项的请求体(带校验规则)"""
    title: str = Field(..., min_length=1, max_length=100, description="标题,1-100 字符")
    description: Optional[str] = Field(None, max_length=500, description="描述,最多 500 字符")

    @validator("title")
    def title_not_blank(cls, v):
    """标题不能为纯空白字符"""
    if not v.strip():
    raise ValueError("标题不能为空")
    return v.strip()

    9.3.2 防止 SQL 注入

    使用 SQLAlchemy 的 ORM 查询会自动参数化,避免 SQL 注入。切勿使用字符串拼接 SQL:

    # ❌ 危险:字符串拼接 SQL,存在注入风险
    # db.execute(f"SELECT * FROM todos WHERE title = '{title}'")

    # ✅ 安全:使用 ORM 参数化查询
    db.query(TodoDB).filter(TodoDB.title == title).all()

    9.3.3 防止 XSS 攻击

    对用户输入的内容进行转义,防止恶意脚本注入:

    import html

    # 对用户输入做 HTML 转义
    safe_title = html.escape(user_input_title)

    9.3.4 请求体大小限制

    限制请求体大小,防止超大请求拖垮服务:

    from fastapi import Request, HTTPException

    MAX_BODY_SIZE = 1024 * 1024 # 1MB

    @app.middleware("http")
    async def limit_body_size(request: Request, call_next):
    """限制请求体大小"""
    content_length = request.headers.get("content-length")
    if content_length and int(content_length) > MAX_BODY_SIZE:
    raise HTTPException(status_code=413, detail="请求体过大")
    return await call_next(request)

    9.4 限流与防暴力破解

    限流可以防止恶意攻击和资源滥用。下面使用 slowapi 实现接口限流:

    pip install slowapi

    from slowapi import Limiter
    from slowapi.util import get_remote_address

    # 创建限流器:默认每分钟 60 次请求
    limiter = Limiter(key_func=get_remote_address)

    @app.post("/login")
    @limiter.limit("5/minute") # 登录接口限制每分钟 5 次,防暴力破解
    def login(request: Request, user: UserLogin):
    """用户登录(带限流)"""
    # 登录逻辑…
    pass

    @app.get("/todos")
    @limiter.limit("100/minute") # 普通接口限制每分钟 100 次
    def list_todos(request: Request, completed: Optional[bool] = None):
    """获取待办事项列表(带限流)"""
    # 查询逻辑…
    pass

    限流策略建议:登录、注册等敏感接口设置较严格的限制(如 5-10 次/分钟);普通查询接口可放宽(如 100-1000 次/分钟);写操作接口适当收紧。同时可结合 IP 黑名单、验证码等手段增强防护#### 9.4.1 基于 Redis 的分布式限流

    slowapi 默认基于内存限流,在单机部署下够用。但如果 API 部署了多个副本(如 Kubernetes 多 Pod),就需要使用 Redis 实现分布式限流,保证所有实例共享同一套计数。

    安装依赖:

    pip install slowapi redis

    配置 Redis 限流存储:

    from slowapi import Limiter
    from slowapi.util import get_remote_address
    from slowapi.middleware import SlowAPIMiddleware
    import redis

    # 连接 Redis(生产环境通过环境变量注入连接串)
    redis_client = redis.Redis(
    host="localhost",
    port=6379,
    db=0,
    decode_responses=True,
    )

    class RedisStorage:
    """基于 Redis 的限流计数存储"""

    def __init__(self, client: redis.Redis):
    self.client = client

    def get(self, key: str) > int:
    """获取当前计数"""
    value = self.client.get(key)
    return int(value) if value else 0

    def incr(self, key: str, expiry: int) > int:
    """计数 +1,并设置过期时间"""
    pipe = self.client.pipeline()
    pipe.incr(key)
    pipe.expire(key, expiry)
    results = pipe.execute()
    return results[0]

    # 使用 Redis 存储创建限流器
    limiter = Limiter(
    key_func=get_remote_address,
    storage_uri="redis://localhost:6379/0",
    )

    @app.post("/login")
    @limiter.limit("5/minute")
    def login(request: Request, user: UserLogin):
    """用户登录(分布式限流,防暴力破解)"""
    # 登录逻辑…
    pass

    @app.get("/todos")
    @limiter.limit("100/minute")
    def list_todos(request: Request, completed: Optional[bool] = None):
    """获取待办事项列表(分布式限流)"""
    # 查询逻辑…
    pass

    # 注册限流中间件
    app.add_middleware(SlowAPIMiddleware)

    分布式限流要点:所有 API 实例共享同一个 Redis 计数,避免单机限流被绕过;expire 设置窗口过期时间,防止计数无限增长;生产环境建议使用 Redis 集群或云厂商托管实例,并配置连接池与超时。

    9.4.2 登录失败锁定策略

    除了限流,还可以在连续多次登录失败后临时锁定账号,进一步降低暴力破解风险:

    from datetime import datetime, timedelta

    # 记录登录失败次数和锁定时间(生产环境应存入 Redis 或数据库)
    login_attempts = {}
    LOCK_THRESHOLD = 5 # 连续失败 5 次锁定
    LOCK_DURATION = timedelta(minutes=15) # 锁定 15 分钟

    def is_account_locked(username: str) > bool:
    """检查账号是否被锁定"""
    record = login_attempts.get(username)
    if not record:
    return False
    if record["locked_until"] and datetime.utcnow() < record["locked_until"]:
    return True
    # 锁定时间已过,重置计数
    if record["locked_until"]:
    login_attempts[username] = {"count": 0, "locked_until": None}
    return False

    @app.post("/login")
    def login(user: UserLogin):
    """用户登录(带失败锁定)"""
    if is_account_locked(user.username):
    raise HTTPException(status_code=423, detail="账号已锁定,请 15 分钟后再试")

    record = fake_users_db.get(user.username)
    if not record or not verify_password(user.password, record["password"]):
    # 记录失败次数
    attempt = login_attempts.get(user.username, {"count": 0, "locked_until": None})
    attempt["count"] += 1
    if attempt["count"] >= LOCK_THRESHOLD:
    attempt["locked_until"] = datetime.utcnow() + LOCK_DURATION
    attempt["count"] = 0
    login_attempts[user.username] = attempt
    raise HTTPException(status_code=401, detail="用户名或密码错误")

    # 登录成功,重置失败计数
    login_attempts[user.username] = {"count": 0, "locked_until": None}
    token = create_token(record["id"])
    return {"access_token": token, "token_type": "bearer"}

    锁定策略要点:连续失败达到阈值后锁定账号,锁定期间即使密码正确也拒绝登录;锁定时间过后自动解锁并重置计数;生产环境建议将计数和锁定状态存入 Redis,并配合验证码进一步增加攻击成本。 。

    9.5 安全最佳实践清单

    • 使用 HTTPS 加密传输
    • 密码使用 bcrypt 等慢哈希算法存储
    • JWT 密钥通过环境变量注入,定期轮换
    • 设置合理的 token 过期时间(建议 15 分钟 – 2 小时)
    • 对敏感接口实施限流
    • 输入数据严格校验,防止注入攻击
    • 日志中不记录密码、token 等敏感信息
    • 定期更新依赖,修复已知漏洞
    • 使用安全头(如 X-Content-Type-Options、X-Frame-Options)
    • 对错误信息做脱敏处理,不泄露内部细节
    9.5.1 安全头配置示例

    在 FastAPI 中通过中间件统一添加安全响应头:

    from starlette.middleware.base import BaseHTTPMiddleware

    class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    """为所有响应添加安全头"""
    async def dispatch(self, request, call_next):
    response = await call_next(request)
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    response.headers["Referrer-Policy"] = "no-referrer"
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
    return response

    app.add_middleware(SecurityHeadersMiddleware)

    9.5.2 敏感信息脱敏示例

    日志和错误响应中避免泄露敏感信息:

    import re

    def mask_sensitive(data: dict) > dict:
    """对敏感字段做脱敏处理"""
    sensitive_fields = {"password", "token", "authorization", "secret"}
    masked = {}
    for key, value in data.items():
    if key.lower() in sensitive_fields:
    masked[key] = "***"
    elif isinstance(value, dict):
    masked[key] = mask_sensitive(value)
    else:
    masked[key] = value
    return masked

    # 使用示例:记录请求日志前先脱敏
    # logger.info(mask_sensitive(request_body))

    安全是持续的过程:没有绝对安全的系统,只有不断加固的系统。建议定期进行安全审计,关注 OWASP Top 10 等安全标准,持续提升 API 的安全性。

    9.6 HTTPS 与 CORS 安全配置

    9.6.1 启用 HTTPS

    生产环境必须使用 HTTPS 加密传输,防止数据在传输过程中被窃听或篡改。在 FastAPI 中,通常由反向代理(Nginx / Caddy)或云平台负载均衡器终止 TLS:

    # Nginx 配置示例
    server {
    listen 443 ssl;
    server_name api.example.com;

    ssl_certificate /etc/ssl/certs/api.example.com.crt;
    ssl_certificate_key /etc/ssl/private/api.example.com.key;
    ssl_protocols TLSv1.2 TLSv1.3;

    location / {
    proxy_pass http://127.0.0.1:8000;
    proxy_set_header Host $host;
    proxy_set_header X-Forwarded-Proto $scheme;
    proxy_set_header X-Real-IP $remote_addr;
    }
    }

    9.6.2 配置 CORS 白名单

    CORS(跨域资源共享)控制哪些前端域名可以访问你的 API。默认情况下 FastAPI 不允许跨域请求,需要显式配置白名单:

    from fastapi.middleware.cors import CORSMiddleware

    # 允许的前端域名白名单(生产环境务必收紧)
    ALLOWED_ORIGINS = [
    "https://my-app.example.com",
    "https://admin.example.com",
    ]

    app.add_middleware(
    CORSMiddleware,
    allow_origins=ALLOWED_ORIGINS, # 生产环境不要用 ["*"]
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
    allow_headers=["Authorization", "Content-Type"],
    )

    CORS 配置要点:生产环境 allow_origins 务必填写具体域名,不要使用 ["*"];如需携带 Cookie 或 Authorization 头,必须设置 allow_credentials=True,此时 allow_origins 不能为通配符。

    9.6.3 完整安全配置示例

    将前面介绍的安全措施整合到一个完整的配置模块中,方便直接复制使用:

    # security_config.py
    """
    API 安全配置整合模块
    包含:安全头、CORS、请求体限制、限流、敏感信息脱敏
    """

    from fastapi import FastAPI, Request, HTTPException
    from fastapi.middleware.cors import CORSMiddleware
    from starlette.middleware.base import BaseHTTPMiddleware
    from slowapi import Limiter
    from slowapi.util import get_remote_address
    from slowapi.middleware import SlowAPIMiddleware
    from slowapi.errors import RateLimitExceeded
    from fastapi.responses import JSONResponse
    import re

    # ========== 1. 限流配置 ==========
    limiter = Limiter(key_func=get_remote_address)

    # ========== 2. 安全响应头中间件 ==========
    class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    """为所有响应添加安全头"""

    async def dispatch(self, request, call_next):
    response = await call_next(request)
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    response.headers["Referrer-Policy"] = "no-referrer"
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
    response.headers["Permissions-Policy"] = "geolocation=(), microphone=(), camera=()"
    return response

    # ========== 3. 请求体大小限制中间件 ==========
    class BodySizeLimitMiddleware(BaseHTTPMiddleware):
    """限制请求体大小,防止超大请求拖垮服务"""

    def __init__(self, app, max_size: int = 1024 * 1024):
    super().__init__(app)
    self.max_size = max_size

    async def dispatch(self, request, call_next):
    content_length = request.headers.get("content-length")
    if content_length and int(content_length) > self.max_size:
    return JSONResponse(
    status_code=413,
    content={"code": 413, "message": "请求体过大", "data": None},
    )
    return await call_next(request)

    # ========== 4. 敏感信息脱敏工具 ==========
    def mask_sensitive(data: dict) > dict:
    """对敏感字段做脱敏处理"""
    sensitive_fields = {"password", "token", "authorization", "secret", "api_key"}
    masked = {}
    for key, value in data.items():
    if key.lower() in sensitive_fields:
    masked[key] = "***"
    elif isinstance(value, dict):
    masked[key] = mask_sensitive(value)
    else:
    masked[key] = value
    return masked

    # ========== 5. 应用安全配置 ==========
    def setup_security(app: FastAPI):
    """为 FastAPI 应用统一配置安全措施"""

    # 5.1 安全响应头
    app.add_middleware(SecurityHeadersMiddleware)

    # 5.2 请求体大小限制(1MB)
    app.add_middleware(BodySizeLimitMiddleware, max_size=1024 * 1024)

    # 5.3 CORS 白名单
    app.add_middleware(
    CORSMiddleware,
    allow_origins=[
    "https://my-app.example.com",
    "https://admin.example.com",
    ],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
    allow_headers=["Authorization", "Content-Type"],
    )

    # 5.4 限流中间件
    app.add_middleware(SlowAPIMiddleware)

    # 5.5 限流异常处理
    @app.exception_handler(RateLimitExceeded)
    async def rate_limit_handler(request: Request, exc: RateLimitExceeded):
    return JSONResponse(
    status_code=429,
    content={"code": 429, "message": "请求过于频繁,请稍后再试", "data": None},
    )

    return app

    # ========== 6. 使用示例 ==========
    app = FastAPI(title="安全配置示例")
    setup_security(app)

    @app.get("/")
    @limiter.limit("60/minute")
    def read_root(request: Request):
    """普通接口:每分钟 60 次"""
    return {"message": "Hello World"}

    @app.post("/login")
    @limiter.limit("5/minute")
    def login(request: Request):
    """登录接口:每分钟 5 次,防暴力破解"""
    # 登录逻辑…
    return {"message": "登录成功"}

    整合要点:将安全配置封装为 setup_security(app) 函数,在应用启动时统一调用,避免散落在各处;限流异常统一返回 429 状态码和友好提示;生产环境请将 SECRET_KEY、数据库连接串等通过环境变量注入,不要硬编码在代码中。

    9.6.4 安全配置实战测试

    配置好安全措施后,我们需要验证它们是否真正生效。下面给出一个完整的安全测试脚本,覆盖认证、授权、限流、输入校验等关键场景:

    # test_security.py
    """
    API 安全配置测试
    验证认证、授权、限流、输入校验等安全措施是否生效
    """

    import requests
    import pytest

    BASE_URL = "http://127.0.0.1:8000"

    @pytest.fixture(scope="module")
    def auth_token():
    """注册并登录,返回有效的 JWT token"""
    # 注册新用户
    username = "test_user"
    password = "Test@12345"

    requests.post(
    f"{BASE_URL}/register",
    json={"username": username, "password": password},
    )

    # 登录获取 token
    resp = requests.post(
    f"{BASE_URL}/login",
    json={"username": username, "password": password},
    )
    assert resp.status_code == 200
    return resp.json()["access_token"]

    def test_register_duplicate_user():
    """重复注册应返回 400"""
    resp = requests.post(
    f"{BASE_URL}/register",
    json={"username": "test_user", "password": "Test@12345"},
    )
    assert resp.status_code == 400

    def test_login_wrong_password():
    """错误密码登录应返回 401"""
    resp = requests.post(
    f"{BASE_URL}/login",
    json={"username": "test_user", "password": "wrong_password"},
    )
    assert resp.status_code == 401

    def test_access_protected_without_token():
    """未携带 token 访问受保护接口应返回 401"""
    resp = requests.get(f"{BASE_URL}/me")
    assert resp.status_code == 401

    def test_access_protected_with_token(auth_token):
    """携带有效 token 访问受保护接口应成功"""
    resp = requests.get(
    f"{BASE_URL}/me",
    headers={"Authorization": f"Bearer {auth_token}"},
    )
    assert resp.status_code == 200

    def test_invalid_token():
    """无效 token 应返回 401"""
    resp = requests.get(
    f"{BASE_URL}/me",
    headers={"Authorization": "Bearer invalid.token.here"},
    )
    assert resp.status_code == 401

    def test_rbac_admin_only():
    """普通用户访问管理员接口应返回 403"""
    # 使用普通用户 token
    resp = requests.post(
    f"{BASE_URL}/login",
    json={"username": "test_user", "password": "Test@12345"},
    )
    user_token = resp.json()["access_token"]

    resp = requests.delete(
    f"{BASE_URL}/todos/1",
    headers={"Authorization": f"Bearer {user_token}"},
    )
    assert resp.status_code == 403

    def test_input_validation():
    """空标题创建待办事项应返回 422"""
    resp = requests.post(
    f"{BASE_URL}/todos",
    json={"title": "", "description": "空标题测试"},
    )
    assert resp.status_code == 422
    data = resp.json()
    assert data["code"] == 422

    def test_body_size_limit():
    """超大请求体应返回 413"""
    big_payload = {"title": "x" * (1024 * 1024 + 100)} # 超过 1MB
    resp = requests.post(f"{BASE_URL}/todos", json=big_payload)
    assert resp.status_code == 413

    def test_rate_limit():
    """连续快速请求应触发限流返回 429"""
    for _ in range(6): # 超过 5 次/分钟的限制
    requests.post(
    f"{BASE_URL}/login",
    json={"username": "test_user", "password": "Test@12345"},
    )
    resp = requests.post(
    f"{BASE_URL}/login",
    json={"username": "test_user", "password": "Test@12345"},
    )
    assert resp.status_code == 429

    def test_security_headers():
    """响应应包含安全头"""
    resp = requests.get(f"{BASE_URL}/")
    assert resp.headers.get("X-Content-Type-Options") == "nosniff"
    assert resp.headers.get("X-Frame-Options") == "DENY"
    assert resp.headers.get("X-XSS-Protection") == "1; mode=block"

    运行方式:

    pytest test_security.py -v

    安全测试要点:测试覆盖认证(401)、授权(403)、输入校验(422)、请求体限制(413)、限流(429)和安全头等关键场景;建议将安全测试纳入 CI/CD 流水线,每次发布前自动执行;生产环境测试时注意使用测试账号,避免影响真实数据。

    9.6.5 SQL 注入防护测试

    SQL 注入是最常见的 Web 安全漏洞之一。虽然 SQLAlchemy ORM 默认使用参数化查询,能有效防止注入,但我们仍应编写测试来验证防护措施确实生效。下面给出一个针对待办事项 API 的 SQL 注入防护测试:

    # test_sql_injection.py
    """
    SQL 注入防护测试
    验证 API 对恶意输入的处理是否符合预期
    """

    import requests
    import pytest

    BASE_URL = "http://127.0.0.1:8000"

    def test_sql_injection_in_title():
    """标题中包含 SQL 注入语句应被安全处理"""
    malicious_payloads = [
    {"title": "'; DROP TABLE todos; –", "description": "尝试删除表"},
    {"title": "' OR '1'='1", "description": "尝试绕过查询条件"},
    {"title": "'; UPDATE todos SET title='hacked' WHERE 1=1; –", "description": "尝试篡改数据"},
    {"title": "<script>alert('xss')</script>", "description": "尝试 XSS 注入"},
    ]

    for payload in malicious_payloads:
    # 创建包含恶意内容的待办事项
    resp = requests.post(f"{BASE_URL}/todos", json=payload)
    assert resp.status_code == 201, f"创建失败: {payload['title']}"

    # 验证返回的数据与提交的一致(未被篡改或执行)
    data = resp.json()
    assert data["title"] == payload["title"], "标题被篡改"
    assert data["description"] == payload["description"], "描述被篡改"

    def test_sql_injection_in_query_params():
    """查询参数中包含 SQL 注入语句应被安全处理"""
    # 尝试通过查询参数注入
    resp = requests.get(
    f"{BASE_URL}/todos",
    params={"completed": "'; DROP TABLE todos; –"},
    )
    # 参数类型错误应返回 422,而不是执行注入语句
    assert resp.status_code in (200, 422)

    def test_sql_injection_in_path_param():
    """路径参数中包含 SQL 注入语句应被安全处理"""
    # 尝试通过路径参数注入
    resp = requests.get(f"{BASE_URL}/todos/'; DROP TABLE todos; –")
    # 非数字 ID 应返回 422 或 404,而不是执行注入语句
    assert resp.status_code in (404, 422)

    def test_database_still_works_after_attacks():
    """注入攻击后数据库应仍然正常工作"""
    # 攻击尝试后,正常创建和查询应不受影响
    resp = requests.post(
    f"{BASE_URL}/todos",
    json={"title": "正常任务", "description": "验证数据库未被破坏"},
    )
    assert resp.status_code == 201

    resp = requests.get(f"{BASE_URL}/todos")
    assert resp.status_code == 200
    assert len(resp.json()) > 0, "数据库中没有数据,可能已被破坏"

    运行方式:

    pytest test_sql_injection.py -v

    SQL 注入防护要点:SQLAlchemy ORM 的参数化查询机制会自动转义特殊字符,因此上述恶意输入只会被当作普通字符串存储,不会执行;测试的核心目的是验证「攻击尝试不会破坏数据库」和「正常功能不受影响」;建议将安全测试纳入 CI/CD 流水线,每次发布前自动执行。

    9.6.6 安全加固完整代码示例

    将前面介绍的安全措施整合为一个可直接运行的完整示例,包含认证、授权、限流、输入校验、安全头和 CORS 配置,方便读者对照学习:

    # secure_main.py
    """
    API 安全加固完整示例
    整合:JWT 认证、RBAC 授权、限流、输入校验、安全头、CORS
    """

    from fastapi import FastAPI, Depends, HTTPException, Request
    from fastapi.middleware.cors import CORSMiddleware
    from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
    from pydantic import BaseModel, Field
    from slowapi import Limiter
    from slowapi.util import get_remote_address
    from slowapi.middleware import SlowAPIMiddleware
    from slowapi.errors import RateLimitExceeded
    from fastapi.responses import JSONResponse
    from starlette.middleware.base import BaseHTTPMiddleware
    import jwt
    import bcrypt
    import time

    # ========== 配置 ==========
    SECRET_KEY = "your-secret-key-change-in-production"
    ALGORITHM = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES = 30

    # ========== 限流 ==========
    limiter = Limiter(key_func=get_remote_address)

    # ========== 安全头中间件 ==========
    class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    """为所有响应添加安全头"""

    async def dispatch(self, request, call_next):
    response = await call_next(request)
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    response.headers["Referrer-Policy"] = "no-referrer"
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
    return response

    # ========== 数据模型 ==========
    class UserCreate(BaseModel):
    """注册请求模型"""
    username: str = Field(..., min_length=3, max_length=20, pattern="^[a-zA-Z0-9_]+$")
    password: str = Field(..., min_length=8, max_length=64)

    class UserLogin(BaseModel):
    """登录请求模型"""
    username: str
    password: str

    class TodoCreate(BaseModel):
    """创建待办事项请求模型"""
    title: str = Field(..., min_length=1, max_length=100)
    description: str = Field(None, max_length=500)

    # ========== 模拟用户存储(生产环境替换为数据库) ==========
    users_db = {}
    todos_db = []
    todo_id_counter = 1

    # ========== 工具函数 ==========
    def hash_password(password: str) > str:
    """使用 bcrypt 哈希密码"""
    return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()

    def verify_password(password: str, hashed: str) > bool:
    """验证密码"""
    return bcrypt.checkpw(password.encode(), hashed.encode())

    def create_token(user_id: int) > str:
    """生成 JWT token"""
    payload = {
    "sub": str(user_id),
    "exp": int(time.time()) + ACCESS_TOKEN_EXPIRE_MINUTES * 60,
    "iat": int(time.time()),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

    def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(HTTPBearer())):
    """解析并验证 JWT token,返回当前用户"""
    try:
    payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM])
    user_id = int(payload["sub"])
    user = users_db.get(user_id)
    if not user:
    raise HTTPException(status_code=401, detail="用户不存在")
    return user
    except jwt.ExpiredSignatureError:
    raise HTTPException(status_code=401, detail="token 已过期")
    except jwt.InvalidTokenError:
    raise HTTPException(status_code=401, detail="无效 token")

    def require_admin(user=Depends(get_current_user)):
    """RBAC:仅管理员可访问"""
    if user["role"] != "admin":
    raise HTTPException(status_code=403, detail="需要管理员权限")
    return user

    # ========== 创建应用 ==========
    app = FastAPI(title="安全加固示例 API")

    # 添加中间件(顺序:安全头 → CORS → 限流)
    app.add_middleware(SecurityHeadersMiddleware)
    app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://my-app.example.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
    allow_headers=["Authorization", "Content-Type"],
    )
    app.add_middleware(SlowAPIMiddleware)

    # ========== 限流异常处理 ==========
    @app.exception_handler(RateLimitExceeded)
    async def rate_limit_handler(request: Request, exc: RateLimitExceeded):
    return JSONResponse(
    status_code=429,
    content={"code": 429, "message": "请求过于频繁,请稍后再试", "data": None},
    )

    # ========== 接口 ==========
    @app.post("/register", status_code=201)
    @limiter.limit("10/minute")
    def register(request: Request, user: UserCreate):
    """注册新用户"""
    # 检查用户名是否已存在
    for existing in users_db.values():
    if existing["username"] == user.username:
    raise HTTPException(status_code=400, detail="用户名已存在")

    user_id = len(users_db) + 1
    users_db[user_id] = {
    "id": user_id,
    "username": user.username,
    "password": hash_password(user.password),
    "role": "user",
    }
    return {"id": user_id, "username": user.username}

    @app.post("/login")
    @limiter.limit("5/minute")
    def login(request: Request, user: UserLogin):
    """登录并返回 JWT token"""
    for record in users_db.values():
    if record["username"] == user.username:
    if not verify_password(user.password, record["password"]):
    raise HTTPException(status_code=401, detail="用户名或密码错误")
    token = create_token(record["id"])
    return {"access_token": token, "token_type": "bearer"}
    raise HTTPException(status_code=401, detail="用户名或密码错误")

    @app.get("/me")
    def get_me(user=Depends(get_current_user)):
    """获取当前登录用户信息"""
    return {"id": user["id"], "username": user["username"], "role": user["role"]}

    @app.get("/admin")
    def admin_only(user=Depends(require_admin)):
    """仅管理员可访问的接口"""
    return {"message": "欢迎管理员", "user": user["username"]}

    @app.post("/todos", status_code=201)
    @limiter.limit("60/minute")
    def create_todo(request: Request, todo: TodoCreate, user=Depends(get_current_user)):
    """创建待办事项(需登录)"""
    global todo_id_counter
    new_todo = {
    "id": todo_id_counter,
    "title": todo.title,
    "description": todo.description,
    "completed": False,
    "owner": user["username"],
    }
    todo_id_counter += 1
    todos_db.append(new_todo)
    return new_todo

    @app.get("/todos")
    def list_todos(user=Depends(get_current_user)):
    """查询当前用户的待办事项列表"""
    return [t for t in todos_db if t["owner"] == user["username"]]

    运行方式:

    # 安装依赖
    pip install fastapi uvicorn pyjwt bcrypt slowapi

    # 启动服务
    uvicorn secure_main:app –reload

    安全加固要点:密码使用 bcrypt 慢哈希存储,绝不保存明文;JWT 密钥通过环境变量注入并定期轮换;登录接口限流防止暴力破解;RBAC 通过依赖注入实现细粒度权限控制;所有输入通过 Pydantic 严格校验,防止注入攻击;生产环境务必使用 HTTPS 并配置 CORS 白名单。

    9.6.7 安全配置完整代码示例(生产级
    9.6.8 安全配置实战测试(生产级)

    下面给出针对 9.6.7 生产级安全配置的完整 pytest 测试脚本,覆盖注册、登录、刷新 Token、权限控制、限流等关键安全场景:

    # test_security.py
    """
    生产级安全配置完整测试
    运行方式:pytest test_security.py -v
    """

    import os
    import tempfile
    import pytest
    from fastapi.testclient import TestClient

    # 使用临时数据库
    TEMP_DB = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
    os.environ["DATABASE_URL"] = f"sqlite:///{TEMP_DB.name}"

    from secure_production import app, Base, engine # noqa: E402

    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)

    client = TestClient(app)

    @pytest.fixture(autouse=True)
    def clean_db():
    """每个用例前重建数据表"""
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)
    yield
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)

    def register_user(username="testuser", password="Test1234"):
    """辅助函数:注册用户"""
    return client.post("/register", json={"username": username, "password": password})

    def login_user(username="testuser", password="Test1234"):
    """辅助函数:登录并返回 token"""
    resp = client.post("/login", json={"username": username, "password": password})
    assert resp.status_code == 200
    return resp.json()

    def auth_header(token):
    """构造 Authorization 头"""
    return {"Authorization": f"Bearer {token}"}

    # ========== 注册测试 ==========
    def test_register_success():
    """测试注册成功"""
    resp = register_user()
    assert resp.status_code == 201
    data = resp.json()
    assert data["username"] == "testuser"
    assert "password" not in data # 响应中不应包含密码

    def test_register_duplicate_username():
    """测试重复用户名注册"""
    register_user()
    resp = register_user()
    assert resp.status_code == 400
    assert resp.json()["detail"] == "用户名已存在"

    def test_register_weak_password():
    """测试弱密码被拒绝"""
    resp = register_user(password="weak")
    assert resp.status_code == 422
    data = resp.json()
    assert data["code"] == 422
    # 应包含密码强度校验错误
    assert any("密码" in item["message"] for item in data["data"])

    def test_register_invalid_username():
    """测试非法用户名被拒绝"""
    resp = register_user(username="bad name!")
    assert resp.status_code == 422

    # ========== 登录测试 ==========
    def test_login_success():
    """测试登录成功,返回 access + refresh token"""
    register_user()
    resp = client.post("/login", json={"username": "testuser", "password": "Test1234"})
    assert resp.status_code == 200
    data = resp.json()
    assert "access_token" in data
    assert "refresh_token" in data
    assert data["token_type"] == "bearer"

    def test_login_wrong_password():
    """测试错误密码"""
    register_user()
    resp = client.post("/login", json={"username": "testuser", "password": "Wrong1234"})
    assert resp.status_code == 401
    assert resp.json()["detail"] == "用户名或密码错误"

    def test_login_nonexistent_user():
    """测试不存在的用户"""
    resp = client.post("/login", json={"username": "ghost", "password": "Test1234"})
    assert resp.status_code == 401

    # ========== Token 刷新测试 ==========
    def test_refresh_token_success():
    """测试刷新 token 成功"""
    register_user()
    tokens = login_user()
    resp = client.post("/refresh", json={"refresh_token": tokens["refresh_token"]})
    assert resp.status_code == 200
    data = resp.json()
    assert "access_token" in data
    assert data["token_type"] == "bearer"

    def test_refresh_token_invalid():
    """测试无效刷新 token"""
    resp = client.post("/refresh", json={"refresh_token": "invalid.token.value"})
    assert resp.status_code == 401

    def test_refresh_token_wrong_type():
    """测试用 access token 刷新(应失败)"""
    register_user()
    tokens = login_user()
    resp = client.post("/refresh", json={"refresh_token": tokens["access_token"]})
    assert resp.status_code == 401

    # ========== 受保护接口测试 ==========
    def test_get_me_without_token():
    """测试未携带 token 访问受保护接口"""
    resp = client.get("/me")
    assert resp.status_code == 403 # HTTPBearer 未提供凭证

    def test_get_me_with_valid_token():
    """测试携带有效 token 访问 /me"""
    register_user()
    tokens = login_user()
    resp = client.get("/me", headers=auth_header(tokens["access_token"]))
    assert resp.status_code == 200
    data = resp.json()
    assert data["username"] == "testuser"
    assert data["role"] == "user"

    def test_get_me_with_invalid_token():
    """测试携带无效 token"""
    resp = client.get("/me", headers=auth_header("invalid.token.value"))
    assert resp.status_code == 401

    # ========== RBAC 权限测试 ==========
    def test_admin_requires_admin_role():
    """测试普通用户访问管理员接口被拒绝"""
    register_user()
    tokens = login_user()
    resp = client.get("/admin", headers=auth_header(tokens["access_token"]))
    assert resp.status_code == 403
    assert resp.json()["detail"] == "需要管理员权限"

    # ========== 安全头测试 ==========
    def test_security_headers_present():
    """测试响应包含安全头"""
    resp = client.get("/")
    assert resp.headers.get("X-Content-Type-Options") == "nosniff"
    assert resp.headers.get("X-Frame-Options") == "DENY"
    assert resp.headers.get("X-XSS-Protection") == "1; mode=block"
    assert resp.headers.get("Referrer-Policy") == "no-referrer"
    assert resp.headers.get("Content-Security-Policy") == "default-src 'self'"

    # ========== 限流测试 ==========
    def test_login_rate_limit():
    """测试登录接口限流(5 次/分钟)"""
    register_user()
    # 连续 5 次错误登录
    for _ in range(5):
    client.post("/login", json={"username": "testuser", "password": "Wrong1234"})

    # 第 6 次应触发限流
    resp = client.post("/login", json={"username": "testuser", "password": "Wrong1234"})
    assert resp.status_code == 429
    assert resp.json()["code"] == 429

    运行方式:

    # 安装测试依赖
    pip install pytest httpx

    # 运行全部安全测试
    pytest test_security.py -v

    测试要点:覆盖注册(成功/重复/弱密码/非法用户名)、登录(成功/错误密码/不存在用户)、Token 刷新(成功/无效/类型错误)、受保护接口(无 token/有效/无效)、RBAC 权限、安全头、限流共 7 大类 18 个用例;每个用例独立重建数据库,互不干扰;限流测试依赖 slowapi 的内存计数,注意测试间隔内不要重复运行。

    前面 9.6.6 给出了整合认证、授权、限流、输入校验、安全头和 CORS 的完整示例。下面再补充一个更贴近生产环境的版本,增加数据库用户存储、Refresh Token 轮换、密码强度校验和日志脱敏,方便读者直接对照落地:

    # secure_production.py
    """
    生产级安全配置完整示例
    在 9.6.6 基础上增加:数据库用户存储、Refresh Token、密码强度校验、日志脱敏
    """

    from fastapi import FastAPI, Depends, HTTPException, Request
    from fastapi.middleware.cors import CORSMiddleware
    from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
    from pydantic import BaseModel, Field, field_validator
    from slowapi import Limiter
    from slowapi.util import get_remote_address
    from slowapi.middleware import SlowAPIMiddleware
    from slowapi.errors import RateLimitExceeded
    from fastapi.responses import JSONResponse
    from starlette.middleware.base import BaseHTTPMiddleware
    from sqlalchemy import create_engine, Column, Integer, String, Boolean
    from sqlalchemy.orm import declarative_base, sessionmaker, Session
    import jwt
    import bcrypt
    import re
    import time
    import logging

    # ========== 配置 ==========
    SECRET_KEY = "your-secret-key-change-in-production"
    REFRESH_SECRET_KEY = "your-refresh-secret-key-change-in-production"
    ALGORITHM = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES = 30
    REFRESH_TOKEN_EXPIRE_DAYS = 7

    # ========== 日志(脱敏) ==========
    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger("secure_api")

    def mask_sensitive(data: dict) > dict:
    """对敏感字段进行脱敏,避免日志泄露"""
    masked = dict(data)
    if "password" in masked:
    masked["password"] = "***"
    if "token" in masked:
    masked["token"] = masked["token"][:8] + "***"
    return masked

    # ========== 数据库 ==========
    DATABASE_URL = "sqlite:///./secure_users.db"
    engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    Base = declarative_base()

    class UserDB(Base):
    """用户表"""
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True, nullable=False)
    password_hash = Column(String, nullable=False)
    role = Column(String, default="user")
    is_active = Column(Boolean, default=True)

    Base.metadata.create_all(bind=engine)

    def get_db():
    """获取数据库会话"""
    db = SessionLocal()
    try:
    yield db
    finally:
    db.close()

    # ========== 限流 ==========
    limiter = Limiter(key_func=get_remote_address)

    # ========== 安全头中间件 ==========
    class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    """为所有响应添加安全头"""

    async def dispatch(self, request, call_next):
    response = await call_next(request)
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    response.headers["Referrer-Policy"] = "no-referrer"
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
    response.headers["Content-Security-Policy"] = "default-src 'self'"
    return response

    # ========== 数据模型 ==========
    class UserCreate(BaseModel):
    """注册请求模型(含密码强度校验)"""
    username: str = Field(..., min_length=3, max_length=20, pattern="^[a-zA-Z0-9_]+$")
    password: str = Field(..., min_length=8, max_length=64)

    @field_validator("password")
    @classmethod
    def validate_password_strength(cls, v):
    """密码必须包含大小写字母和数字"""
    if not re.search(r"[a-z]", v):
    raise ValueError("密码必须包含小写字母")
    if not re.search(r"[A-Z]", v):
    raise ValueError("密码必须包含大写字母")
    if not re.search(r"\\d", v):
    raise ValueError("密码必须包含数字")
    return v

    class UserLogin(BaseModel):
    """登录请求模型"""
    username: str
    password: str

    class RefreshTokenRequest(BaseModel):
    """刷新 token 请求模型"""
    refresh_token: str

    # ========== 工具函数 ==========
    def hash_password(password: str) > str:
    """使用 bcrypt 哈希密码"""
    return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()

    def verify_password(password: str, hashed: str) > bool:
    """验证密码"""
    return bcrypt.checkpw(password.encode(), hashed.encode())

    def create_access_token(user_id: int) > str:
    """生成短期访问 token"""
    payload = {
    "sub": str(user_id),
    "type": "access",
    "exp": int(time.time()) + ACCESS_TOKEN_EXPIRE_MINUTES * 60,
    "iat": int(time.time()),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

    def create_refresh_token(user_id: int) > str:
    """生成长效刷新 token"""
    payload = {
    "sub": str(user_id),
    "type": "refresh",
    "exp": int(time.time()) + REFRESH_TOKEN_EXPIRE_DAYS * 86400,
    "iat": int(time.time()),
    }
    return jwt.encode(payload, REFRESH_SECRET_KEY, algorithm=ALGORITHM)

    def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(HTTPBearer()),
    db: Session = Depends(get_db),
    ):
    """解析并验证访问 token,返回当前用户"""
    try:
    payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM])
    if payload.get("type") != "access":
    raise HTTPException(status_code=401, detail="无效 token 类型")
    user = db.query(UserDB).filter(UserDB.id == int(payload["sub"])).first()
    if not user or not user.is_active:
    raise HTTPException(status_code=401, detail="用户不存在或已禁用")
    return user
    except jwt.ExpiredSignatureError:
    raise HTTPException(status_code=401, detail="token 已过期")
    except jwt.InvalidTokenError:
    raise HTTPException(status_code=401, detail="无效 token")

    def require_admin(user=Depends(get_current_user)):
    """RBAC:仅管理员可访问"""
    if user.role != "admin":
    raise HTTPException(status_code=403, detail="需要管理员权限")
    return user

    # ========== 创建应用 ==========
    app = FastAPI(title="生产级安全配置示例")

    # 添加中间件(顺序:安全头 → CORS → 限流)
    app.add_middleware(SecurityHeadersMiddleware)
    app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://my-app.example.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
    allow_headers=["Authorization", "Content-Type"],
    )
    app.add_middleware(SlowAPIMiddleware)

    # ========== 限流异常处理 ==========
    @app.exception_handler(RateLimitExceeded)
    async def rate_limit_handler(request: Request, exc: RateLimitExceeded):
    return JSONResponse(
    status_code=429,
    content={"code": 429, "message": "请求过于频繁,请稍后再试", "data": None},
    )

    # ========== 接口 ==========
    @app.post("/register", status_code=201)
    @limiter.limit("10/minute")
    def register(request: Request, user: UserCreate, db: Session = Depends(get_db)):
    """注册新用户(密码强度已由 Pydantic 校验)"""
    if db.query(UserDB).filter(UserDB.username == user.username).first():
    raise HTTPException(status_code=400, detail="用户名已存在")

    new_user = UserDB(
    username=user.username,
    password_hash=hash_password(user.password),
    role="user",
    )
    db.add(new_user)
    db.commit()
    db.refresh(new_user)

    logger.info("新用户注册: %s", new_user.username)
    return {"id": new_user.id, "username": new_user.username}

    @app.post("/login")
    @limiter.limit("5/minute")
    def login(request: Request, user: UserLogin, db: Session = Depends(get_db)):
    """登录并返回访问 token + 刷新 token"""
    record = db.query(UserDB).filter(UserDB.username == user.username).first()
    if not record or not verify_password(user.password, record.password_hash):
    logger.warning("登录失败: %s", user.username)
    raise HTTPException(status_code=401, detail="用户名或密码错误")

    access_token = create_access_token(record.id)
    refresh_token = create_refresh_token(record.id)
    logger.info("用户登录成功: %s", record.username)
    return {
    "access_token": access_token,
    "refresh_token": refresh_token,
    "token_type": "bearer",
    }

    @app.post("/refresh")
    @limiter.limit("10/minute")
    def refresh_token(request: Request, body: RefreshTokenRequest):
    """使用刷新 token 换取新的访问 token"""
    try:
    payload = jwt.decode(body.refresh_token, REFRESH_SECRET_KEY, algorithms=[ALGORITHM])
    if payload.get("type") != "refresh":
    raise HTTPException(status_code=401, detail="无效 token 类型")
    new_access_token = create_access_token(int(payload["sub"]))
    return {"access_token": new_access_token, "token_type": "bearer"}
    except jwt.ExpiredSignatureError:
    raise HTTPException(status_code=401, detail="刷新 token 已过期,请重新登录")
    except jwt.InvalidTokenError:
    raise HTTPException(status_code=401, detail="无效刷新 token")

    @app.get("/me")
    def get_me(user=Depends(get_current_user)):
    """获取当前登录用户信息"""
    return {"id": user.id, "username": user.username, "role": user.role}

    @app.get("/admin")
    def admin_only(user=Depends(require_admin)):
    """仅管理员可访问的接口"""
    return {"message": "欢迎管理员", "user": user.username}

    运行方式:

    # 安装依赖
    pip install fastapi uvicorn pyjwt bcrypt slowapi sqlalchemy

    # 启动服务
    uvicorn secure_production:app –reload

    生产级安全要点:用户数据存入数据库而非内存,服务重启不丢失;访问 token 短期有效(30 分钟),配合 Refresh Token 轮换机制,降低泄露风险;密码强度由 Pydantic 校验器强制要求大小写字母 + 数字;日志记录时对密码等敏感字段脱敏,避免泄露;安全头增加 Content-Security-Policy,进一步缓解 XSS 攻击。

    9.6.9 安全配置实战测试(生产级,完整可运行
    9.6.10 安全配置完整可运行示例(生产级,含完整依赖)

    在 9.6.9 的基础上,补充一个可直接复制运行的完整安全配置示例,将前面分散的认证、授权、限流、安全头、日志脱敏整合为一个文件,方便读者直接部署到生产环境。

    # secure_production_full.py
    """
    生产级安全配置完整示例
    整合:JWT 认证、RBAC 授权、限流、安全头、日志脱敏、数据库持久化
    可直接运行:uvicorn secure_production_full:app –reload
    """

    import logging
    import re
    import time
    from datetime import datetime, timedelta
    from typing import Optional

    import bcrypt
    import jwt
    from fastapi import Depends, FastAPI, HTTPException, Request
    from fastapi.middleware.cors import CORSMiddleware
    from fastapi.responses import JSONResponse
    from pydantic import BaseModel, Field, field_validator
    from slowapi import Limiter, _rate_limit_exceeded_handler
    from slowapi.errors import RateLimitExceeded
    from slowapi.middleware import SlowAPIMiddleware
    from slowapi.util import get_remote_address
    from sqlalchemy import Boolean, Column, Integer, String, create_engine
    from sqlalchemy.orm import Session, declarative_base, sessionmaker

    # ========== 日志配置(含脱敏) ==========
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
    logger = logging.getLogger("secure_api")

    SENSITIVE_KEYS = {"password", "token", "refresh_token", "authorization", "secret"}

    def mask_sensitive(data: dict) > dict:
    """对敏感字段进行脱敏处理"""
    masked = {}
    for key, value in data.items():
    if key.lower() in SENSITIVE_KEYS and isinstance(value, str):
    if len(value) <= 6:
    masked[key] = "***"
    else:
    masked[key] = value[:6] + "***"
    else:
    masked[key] = value
    return masked

    class SensitiveFilter(logging.Filter):
    """日志过滤器:自动脱敏敏感字段"""

    def filter(self, record):
    if isinstance(record.args, dict):
    record.args = mask_sensitive(record.args)
    elif isinstance(record.args, tuple):
    record.args = tuple(mask_sensitive(a) if isinstance(a, dict) else a for a in record.args)
    return True

    logger.addFilter(SensitiveFilter())

    # ========== 数据库配置 ==========
    DATABASE_URL = "sqlite:///./secure_users.db"
    engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    Base = declarative_base()

    class UserDB(Base):
    """用户表"""
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True, nullable=False)
    password_hash = Column(String, nullable=False)
    role = Column(String, default="user")
    is_active = Column(Boolean, default=True)

    Base.metadata.create_all(bind=engine)

    def get_db():
    db = SessionLocal()
    try:
    yield db
    finally:
    db.close()

    # ========== 密码哈希 ==========
    def hash_password(password: str) > str:
    """使用 bcrypt 哈希密码"""
    return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")

    def verify_password(password: str, password_hash: str) > bool:
    """验证密码"""
    return bcrypt.checkpw(password.encode("utf-8"), password_hash.encode("utf-8"))

    # ========== JWT 配置 ==========
    SECRET_KEY = "your-secret-key-change-in-production"
    REFRESH_SECRET_KEY = "your-refresh-secret-key-change-in-production"
    ALGORITHM = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES = 30
    REFRESH_TOKEN_EXPIRE_DAYS = 7

    def create_access_token(user_id: int) > str:
    """生成访问 token(短期有效)"""
    payload = {
    "sub": str(user_id),
    "type": "access",
    "exp": datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    "iat": datetime.utcnow(),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

    def create_refresh_token(user_id: int) > str:
    """生成刷新 token(长期有效)"""
    payload = {
    "sub": str(user_id),
    "type": "refresh",
    "exp": datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS),
    "iat": datetime.utcnow(),
    }
    return jwt.encode(payload, REFRESH_SECRET_KEY, algorithm=ALGORITHM)

    # ========== 安全头中间件 ==========
    class SecurityHeadersMiddleware:
    """为所有响应添加安全头"""

    def __init__(self, app):
    self.app = app

    async def __call__(self, scope, receive, send):
    if scope["type"] != "http":
    return await self.app(scope, receive, send)

    async def send_wrapper(message):
    if message["type"] == "http.response.start":
    headers = dict(message.get("headers", []))
    security_headers = {
    b"X-Content-Type-Options": b"nosniff",
    b"X-Frame-Options": b"DENY",
    b"X-XSS-Protection": b"1; mode=block",
    b"Referrer-Policy": b"no-referrer",
    b"Strict-Transport-Security": b"max-age=31536000; includeSubDomains",
    b"Content-Security-Policy": b"default-src 'self'",
    }
    headers.update(security_headers)
    message["headers"] = list(headers.items())
    await send(message)

    await self.app(scope, receive, send_wrapper)

    # ========== 限流配置 ==========
    limiter = Limiter(key_func=get_remote_address)

    # ========== Pydantic 模型 ==========
    class UserCreate(BaseModel):
    """注册请求模型"""
    username: str = Field(..., min_length=3, max_length=50, pattern="^[a-zA-Z0-9_]+$")
    password: str = Field(..., min_length=8, max_length=128)

    @field_validator("password")
    @classmethod
    def validate_password_strength(cls, v):
    """密码强度校验:必须包含大小写字母和数字"""
    if not re.search(r"[a-z]", v):
    raise ValueError("密码必须包含小写字母")
    if not re.search(r"[A-Z]", v):
    raise ValueError("密码必须包含大写字母")
    if not re.search(r"\\d", v):
    raise ValueError("密码必须包含数字")
    return v

    class UserLogin(BaseModel):
    """登录请求模型"""
    username: str
    password: str

    class RefreshTokenRequest(BaseModel):
    """刷新 token 请求模型"""
    refresh_token: str

    # ========== 认证依赖 ==========
    def get_current_user(token: str, db: Session = Depends(get_db)):
    """解析 JWT 并返回当前用户"""
    credentials_exception = HTTPException(status_code=401, detail="无效 token")
    try:
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    if payload.get("type") != "access":
    raise credentials_exception
    user = db.query(UserDB).filter(UserDB.id == int(payload["sub"])).first()
    if not user or not user.is_active:
    raise credentials_exception
    return user
    except jwt.ExpiredSignatureError:
    raise HTTPException(status_code=401, detail="token 已过期")
    except jwt.InvalidTokenError:
    raise credentials_exception

    def require_admin(user=Depends(get_current_user)):
    """RBAC:仅管理员可访问"""
    if user.role != "admin":
    raise HTTPException(status_code=403, detail="需要管理员权限")
    return user

    # ========== 创建应用 ==========
    app = FastAPI(title="生产级安全配置完整示例")

    # 添加中间件(顺序:安全头 → CORS → 限流)
    app.add_middleware(SecurityHeadersMiddleware)
    app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://my-app.example.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
    allow_headers=["Authorization", "Content-Type"],
    )
    app.add_middleware(SlowAPIMiddleware)
    app.state.limiter = limiter
    app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

    # ========== 接口 ==========
    @app.post("/register", status_code=201)
    @limiter.limit("10/minute")
    def register(request: Request, user: UserCreate, db: Session = Depends(get_db)):
    """注册新用户"""
    if db.query(UserDB).filter(UserDB.username == user.username).first():
    raise HTTPException(status_code=400, detail="用户名已存在")

    new_user = UserDB(
    username=user.username,
    password_hash=hash_password(user.password),
    role="user",
    )
    db.add(new_user)
    db.commit()
    db.refresh(new_user)

    logger.info("新用户注册: %s", {"username": new_user.username})
    return {"id": new_user.id, "username": new_user.username}

    @app.post("/login")
    @limiter.limit("5/minute")
    def login(request: Request, user: UserLogin, db: Session = Depends(get_db)):
    """登录并返回访问 token + 刷新 token"""
    record = db.query(UserDB).filter(UserDB.username == user.username).first()
    if not record or not verify_password(user.password, record.password_hash):
    logger.warning("登录失败: %s", {"username": user.username})
    raise HTTPException(status_code=401, detail="用户名或密码错误")

    access_token = create_access_token(record.id)
    refresh_token = create_refresh_token(record.id)
    logger.info("用户登录成功: %s", {"username": record.username})
    return {
    "access_token": access_token,
    "refresh_token": refresh_token,
    "token_type": "bearer",
    }

    @app.post("/refresh")
    @limiter.limit("10/minute")
    def refresh_token(request: Request, body: RefreshTokenRequest):
    """使用刷新 token 换取新的访问 token"""
    try:
    payload = jwt.decode(body.refresh_token, REFRESH_SECRET_KEY, algorithms=[ALGORITHM])
    if payload.get("type") != "refresh":
    raise HTTPException(status_code=401, detail="无效 token 类型")
    new_access_token = create_access_token(int(payload["sub"]))
    return {"access_token": new_access_token, "token_type": "bearer"}
    except jwt.ExpiredSignatureError:
    raise HTTPException(status_code=401, detail="刷新 token 已过期,请重新登录")
    except jwt.InvalidTokenError:
    raise HTTPException(status_code=401, detail="无效刷新 token")

    @app.get("/me")
    def get_me(user=Depends(get_current_user)):
    """获取当前登录用户信息"""
    return {"id": user.id, "username": user.username, "role": user.role}

    @app.get("/admin")
    def admin_only(user=Depends(require_admin)):
    """仅管理员可访问的接口"""
    return {"message": "欢迎管理员", "user": user.username}

    @app.get("/health")
    def health_check():
    """健康检查接口"""
    return {"status": "ok", "time": datetime.utcnow().isoformat()}

    配套的 requirements.txt:

    fastapi==0.115.0
    uvicorn[standard]==0.30.6
    pyjwt==2.9.0
    bcrypt==4.2.0
    slowapi==0.1.9
    sqlalchemy==2.0.35
    pydantic==2.9.2
    pytest==8.3.3
    httpx==0.27.2

    运行方式:

    # 安装依赖
    pip install -r requirements.txt

    # 启动服务
    uvicorn secure_production_full:app –reload

    # 验证安全头
    curl -sI http://127.0.0.1:8000/health | grep -i "x-content-type\\|x-frame\\|strict-transport"

    # 验证限流(连续请求 6 次登录接口)
    for i in $(seq 1 6); do
    curl -s -o /dev/null -w "%{http_code}\\n" -X POST http://127.0.0.1:8000/login \\
    -H "Content-Type: application/json" \\
    -d '{"username": "test", "password": "Test1234"}'
    done

    完整示例要点:本文件将 9.1 的 JWT 认证、9.2 的 RBAC、9.4 的限流、9.5 的安全头与日志脱敏、9.6 的 CORS 与 HTTPS 配置整合为一个可直接运行的生产级示例;requirements.txt 锁定了所有依赖版本,避免因版本升级导致行为变化;/health 接口供 Docker 健康检查使用;限流测试脚本可直观验证 5 次/分钟的登录限制。

    在 9.6.8 的基础上,补充针对生产级安全配置(secure_production.py)的完整 pytest 测试脚本,覆盖数据库用户存储、Refresh Token 轮换、密码强度校验、日志脱敏等新增特性。

    # test_secure_production.py
    """
    生产级安全配置测试
    覆盖:注册(密码强度/重复用户名)、登录、Refresh Token 轮换、
    RBAC、安全头、日志脱敏、数据库持久化
    """

    import io
    import logging
    import pytest
    from fastapi.testclient import TestClient
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.pool import StaticPool

    from secure_production import (
    app,
    Base,
    get_db,
    UserDB,
    hash_password,
    verify_password,
    create_access_token,
    create_refresh_token,
    )

    # 使用内存数据库
    engine = create_engine(
    "sqlite://",
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
    )
    TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

    def override_get_db():
    db = TestingSessionLocal()
    try:
    yield db
    finally:
    db.close()

    app.dependency_overrides[get_db] = override_get_db
    client = TestClient(app)

    @pytest.fixture(autouse=True)
    def setup_db():
    """每个测试前重建数据表"""
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)
    yield

    def register_user(username="testuser", password="Test1234"):
    """辅助函数:注册用户"""
    return client.post("/register", json={"username": username, "password": password})

    def login_user(username="testuser", password="Test1234"):
    """辅助函数:登录并返回 token"""
    resp = client.post("/login", json={"username": username, "password": password})
    assert resp.status_code == 200
    return resp.json()

    def auth_header(token):
    """构造 Authorization 头"""
    return {"Authorization": f"Bearer {token}"}

    # ========== 密码强度校验 ==========
    def test_register_weak_password_no_lowercase():
    """密码缺少小写字母"""
    resp = register_user("user1", "TEST1234")
    assert resp.status_code == 422
    assert "小写字母" in str(resp.json())

    def test_register_weak_password_no_uppercase():
    """密码缺少大写字母"""
    resp = register_user("user2", "test1234")
    assert resp.status_code == 422
    assert "大写字母" in str(resp.json())

    def test_register_weak_password_no_digit():
    """密码缺少数字"""
    resp = register_user("user3", "Testabcd")
    assert resp.status_code == 422
    assert "数字" in str(resp.json())

    def test_register_short_password():
    """密码长度不足 8 位"""
    resp = register_user("user4", "Test12")
    assert resp.status_code == 422

    def test_register_invalid_username():
    """用户名含非法字符"""
    resp = register_user("bad user!", "Test1234")
    assert resp.status_code == 422

    # ========== 注册与重复用户名 ==========
    def test_register_success():
    """正常注册返回 201"""
    resp = register_user()
    assert resp.status_code == 201
    data = resp.json()
    assert data["username"] == "testuser"
    assert "password" not in data # 不返回密码

    def test_register_duplicate_username():
    """重复用户名返回 400"""
    register_user()
    resp = register_user()
    assert resp.status_code == 400
    assert resp.json()["detail"] == "用户名已存在"

    # ========== 密码哈希 ==========
    def test_password_hash_and_verify():
    """密码哈希与验证"""
    hashed = hash_password("Test1234")
    assert hashed != "Test1234" # 不存明文
    assert verify_password("Test1234", hashed)
    assert not verify_password("Wrong1234", hashed)

    def test_password_hash_is_random():
    """同一密码两次哈希结果不同(加盐)"""
    h1 = hash_password("Test1234")
    h2 = hash_password("Test1234")
    assert h1 != h2

    # ========== 登录 ==========
    def test_login_success_returns_both_tokens():
    """登录成功返回 access + refresh token"""
    register_user()
    data = login_user()
    assert "access_token" in data
    assert "refresh_token" in data
    assert data["token_type"] == "bearer"

    def test_login_wrong_password():
    """错误密码返回 401"""
    register_user()
    resp = client.post("/login", json={"username": "testuser", "password": "Wrong1234"})
    assert resp.status_code == 401

    def test_login_nonexistent_user():
    """不存在的用户返回 401"""
    resp = client.post("/login", json={"username": "ghost", "password": "Test1234"})
    assert resp.status_code == 401

    # ========== Refresh Token 轮换 ==========
    def test_refresh_token_success():
    """刷新 token 成功返回新 access token"""
    register_user()
    tokens = login_user()
    resp = client.post("/refresh", json={"refresh_token": tokens["refresh_token"]})
    assert resp.status_code == 200
    data = resp.json()
    assert "access_token" in data
    assert data["token_type"] == "bearer"

    def test_refresh_token_invalid():
    """无效刷新 token 返回 401"""
    resp = client.post("/refresh", json={"refresh_token": "invalid.token.value"})
    assert resp.status_code == 401

    def test_refresh_token_wrong_type():
    """用 access token 刷新应失败"""
    register_user()
    tokens = login_user()
    resp = client.post("/refresh", json={"refresh_token": tokens["access_token"]})
    assert resp.status_code == 401

    def test_refresh_token_expired():
    """过期刷新 token 返回 401"""
    import time
    import jwt
    from secure_production import REFRESH_SECRET_KEY, ALGORITHM

    # 构造一个已过期的 refresh token
    expired_payload = {
    "sub": "1",
    "type": "refresh",
    "exp": int(time.time()) 3600,
    "iat": int(time.time()) 7200,
    }
    expired_token = jwt.encode(expired_payload, REFRESH_SECRET_KEY, algorithm=ALGORITHM)
    resp = client.post("/refresh", json={"refresh_token": expired_token})
    assert resp.status_code == 401
    assert "过期" in resp.json()["detail"]

    # ========== 受保护接口 ==========
    def test_get_me_without_token():
    """未携带 token 返回 403"""
    resp = client.get("/me")
    assert resp.status_code == 403

    def test_get_me_with_valid_token():
    """携带有效 token 返回用户信息"""
    register_user()
    tokens = login_user()
    resp = client.get("/me", headers=auth_header(tokens["access_token"]))
    assert resp.status_code == 200
    data = resp.json()
    assert data["username"] == "testuser"
    assert data["role"] == "user"

    def test_get_me_with_invalid_token():
    """无效 token 返回 401"""
    resp = client.get("/me", headers=auth_header("invalid.token.value"))
    assert resp.status_code == 401

    # ========== RBAC 权限 ==========
    def test_admin_requires_admin_role():
    """普通用户访问管理员接口被拒绝"""
    register_user()
    tokens = login_user()
    resp = client.get("/admin", headers=auth_header(tokens["access_token"]))
    assert resp.status_code == 403
    assert resp.json()["detail"] == "需要管理员权限"

    # ========== 安全头 ==========
    def test_security_headers_present():
    """响应包含完整安全头"""
    resp = client.get("/")
    assert resp.headers.get("X-Content-Type-Options") == "nosniff"
    assert resp.headers.get("X-Frame-Options") == "DENY"
    assert resp.headers.get("X-XSS-Protection") == "1; mode=block"
    assert resp.headers.get("Referrer-Policy") == "no-referrer"
    assert resp.headers.get("Strict-Transport-Security") == "max-age=31536000; includeSubDomains"
    assert resp.headers.get("Content-Security-Policy") == "default-src 'self'"

    # ========== 数据库持久化 ==========
    def test_user_persisted_in_db():
    """注册后用户写入数据库"""
    register_user()
    db = TestingSessionLocal()
    user = db.query(UserDB).filter(UserDB.username == "testuser").first()
    db.close()
    assert user is not None
    assert user.role == "user"
    assert user.is_active is True
    assert user.password_hash != "Test1234" # 存的是哈希

    def test_user_password_hash_verifies():
    """数据库中的哈希能正确验证密码"""
    register_user()
    db = TestingSessionLocal()
    user = db.query(UserDB).filter(UserDB.username == "testuser").first()
    db.close()
    assert verify_password("Test1234", user.password_hash)
    assert not verify_password("Wrong1234", user.password_hash)

    # ========== 日志脱敏 ==========
    def test_log_masking():
    """日志中密码被脱敏"""
    from secure_production import mask_sensitive

    masked = mask_sensitive({"username": "testuser", "password": "Test1234"})
    assert masked["password"] == "***"
    assert masked["username"] == "testuser"

    def test_log_masking_token():
    """日志中 token 被部分脱敏"""
    from secure_production import mask_sensitive

    masked = mask_sensitive({"token": "abcdefghijklmnop"})
    assert masked["token"] == "abcdefgh***"
    assert "ijklmnop" not in masked["token"]

    运行方式:

    # 安装依赖
    pip install fastapi uvicorn pyjwt bcrypt slowapi sqlalchemy pytest httpx

    # 运行生产级安全测试
    pytest test_secure_production.py -v

    测试要点:覆盖密码强度(缺小写/大写/数字/长度不足/非法用户名)、注册(成功/重复/不返回密码)、密码哈希(加盐随机/可验证)、登录(成功/错误密码/不存在用户)、Refresh Token(成功/无效/类型错误/过期)、受保护接口(无 token/有效/无效)、RBAC、安全头、数据库持久化、日志脱敏共 10 大类 24 个用例;每个用例独立重建内存数据库,互不干扰;mask_sensitive 函数可单独测试,确保日志不会泄露敏感信息。

    9.6.11 安全配置验证测试脚本(完整可运行)

    下面提供一个针对生产级安全配置(secure_production_full.py)的完整 pytest 测试脚本,覆盖认证、RBAC、限流、安全头、日志脱敏等核心安全特性,方便你在改动安全配置后一键回归验证。

    # test_security_full.py
    """
    生产级安全配置完整测试
    覆盖:注册、登录、JWT 认证、RBAC、限流、安全头、日志脱敏
    """

    import io
    import logging
    import time

    import jwt
    import pytest
    from fastapi.testclient import TestClient
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.pool import StaticPool

    from secure_production_full import (
    app,
    Base,
    get_db,
    UserDB,
    hash_password,
    verify_password,
    create_access_token,
    create_refresh_token,
    REFRESH_SECRET_KEY,
    ALGORITHM,
    mask_sensitive,
    )

    # 使用独立的内存数据库,避免污染真实数据
    engine = create_engine(
    "sqlite://",
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
    )
    TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

    def override_get_db():
    db = TestingSessionLocal()
    try:
    yield db
    finally:
    db.close()

    app.dependency_overrides[get_db] = override_get_db
    client = TestClient(app)

    @pytest.fixture(autouse=True)
    def setup_db():
    """每个测试前重建数据表"""
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)
    yield

    def register_user(username="testuser", password="Test1234"):
    """辅助函数:注册用户"""
    return client.post("/register", json={"username": username, "password": password})

    def login_user(username="testuser", password="Test1234"):
    """辅助函数:登录并返回 token"""
    resp = client.post("/login", json={"username": username, "password": password})
    assert resp.status_code == 200
    return resp.json()

    def auth_header(token):
    """构造 Authorization 头"""
    return {"Authorization": f"Bearer {token}"}

    # ========== 注册与密码强度 ==========
    def test_register_success():
    """正常注册返回 201,且不返回密码"""
    resp = register_user()
    assert resp.status_code == 201
    data = resp.json()
    assert data["username"] == "testuser"
    assert "password" not in data

    def test_register_duplicate_username():
    """重复用户名返回 400"""
    register_user()
    resp = register_user()
    assert resp.status_code == 400
    assert resp.json()["detail"] == "用户名已存在"

    def test_register_weak_password_no_lowercase():
    """密码缺少小写字母返回 422"""
    resp = register_user("user1", "TEST1234")
    assert resp.status_code == 422

    def test_register_weak_password_no_uppercase():
    """密码缺少大写字母返回 422"""
    resp = register_user("user2", "test1234")
    assert resp.status_code == 422

    def test_register_weak_password_no_digit():
    """密码缺少数字返回 422"""
    resp = register_user("user3", "Testabcd")
    assert resp.status_code == 422

    def test_register_short_password():
    """密码长度不足 8 位返回 422"""
    resp = register_user("user4", "Test12")
    assert resp.status_code == 422

    # ========== 密码哈希 ==========
    def test_password_hash_and_verify():
    """密码哈希可验证且不存明文"""
    hashed = hash_password("Test1234")
    assert hashed != "Test1234"
    assert verify_password("Test1234", hashed)
    assert not verify_password("Wrong1234", hashed)

    def test_password_hash_is_random():
    """同一密码两次哈希结果不同(加盐)"""
    h1 = hash_password("Test1234")
    h2 = hash_password("Test1234")
    assert h1 != h2

    # ========== 登录 ==========
    def test_login_success_returns_both_tokens():
    """登录成功返回 access + refresh token"""
    register_user()
    data = login_user()
    assert "access_token" in data
    assert "refresh_token" in data
    assert data["token_type"] == "bearer"

    def test_login_wrong_password():
    """错误密码返回 401"""
    register_user()
    resp = client.post("/login", json={"username": "testuser", "password": "Wrong1234"})
    assert resp.status_code == 401

    def test_login_nonexistent_user():
    """不存在的用户返回 401"""
    resp = client.post("/login", json={"username": "ghost", "password": "Test1234"})
    assert resp.status_code == 401

    # ========== Refresh Token 轮换 ==========
    def test_refresh_token_success():
    """刷新 token 成功返回新 access token"""
    register_user()
    tokens = login_user()
    resp = client.post("/refresh", json={"refresh_token": tokens["refresh_token"]})
    assert resp.status_code == 200
    assert "access_token" in resp.json()

    def test_refresh_token_invalid():
    """无效刷新 token 返回 401"""
    resp = client.post("/refresh", json={"refresh_token": "invalid.token.value"})
    assert resp.status_code == 401

    def test_refresh_token_wrong_type():
    """用 access token 刷新应失败"""
    register_user()
    tokens = login_user()
    resp = client.post("/refresh", json={"refresh_token": tokens["access_token"]})
    assert resp.status_code == 401

    def test_refresh_token_expired():
    """过期刷新 token 返回 401"""
    expired_payload = {
    "sub": "1",
    "type": "refresh",
    "exp": int(time.time()) 3600,
    "iat": int(time.time()) 7200,
    }
    expired_token = jwt.encode(expired_payload, REFRESH_SECRET_KEY, algorithm=ALGORITHM)
    resp = client.post("/refresh", json={"refresh_token": expired_token})
    assert resp.status_code == 401
    assert "过期" in resp.json()["detail"]

    # ========== 受保护接口与 RBAC ==========
    def test_get_me_without_token():
    """未携带 token 返回 403"""
    resp = client.get("/me")
    assert resp.status_code == 403

    def test_get_me_with_valid_token():
    """携带有效 token 返回用户信息"""
    register_user()
    tokens = login_user()
    resp = client.get("/me", headers=auth_header(tokens["access_token"]))
    assert resp.status_code == 200
    assert resp.json()["username"] == "testuser"

    def test_get_me_with_invalid_token():
    """无效 token 返回 401"""
    resp = client.get("/me", headers=auth_header("invalid.token.value"))
    assert resp.status_code == 401

    def test_admin_requires_admin_role():
    """普通用户访问管理员接口被拒绝"""
    register_user()
    tokens = login_user()
    resp = client.get("/admin", headers=auth_header(tokens["access_token"]))
    assert resp.status_code == 403
    assert resp.json()["detail"] == "需要管理员权限"

    # ========== 安全头 ==========
    def test_security_headers_present():
    """响应包含完整安全头"""
    resp = client.get("/health")
    assert resp.headers.get("X-Content-Type-Options") == "nosniff"
    assert resp.headers.get("X-Frame-Options") == "DENY"
    assert resp.headers.get("X-XSS-Protection") == "1; mode=block"
    assert resp.headers.get("Referrer-Policy") == "no-referrer"
    assert resp.headers.get("Strict-Transport-Security") == "max-age=31536000; includeSubDomains"
    assert resp.headers.get("Content-Security-Policy") == "default-src 'self'"

    # ========== 数据库持久化 ==========
    def test_user_persisted_in_db():
    """注册后用户写入数据库"""
    register_user()
    db = TestingSessionLocal()
    user = db.query(UserDB).filter(UserDB.username == "testuser").first()
    db.close()
    assert user is not None
    assert user.role == "user"
    assert user.password_hash != "Test1234"

    # ========== 日志脱敏 ==========
    def test_log_masking_password():
    """日志中密码被脱敏"""
    masked = mask_sensitive({"username": "testuser", "password": "Test1234"})
    assert masked["password"] == "***"
    assert masked["username"] == "testuser"

    def test_log_masking_token():
    """日志中 token 被部分脱敏"""
    masked = mask_sensitive({"token": "abcdefghijklmnop"})
    assert masked["token"] == "abcdefgh***"
    assert "ijklmnop" not in masked["token"]

    配套的 requirements.txt(安全测试专用):

    fastapi==0.115.0
    uvicorn[standard]==0.30.6
    pyjwt==2.9.0
    bcrypt==4.2.0
    slowapi==0.1.9
    sqlalchemy==2.0.35
    pydantic==2.9.2
    pytest==8.3.3
    httpx==0.27.2

    运行方式:

    # 安装依赖
    pip install -r requirements.txt

    # 运行安全测试
    pytest test_security_full.py -v

    测试要点:覆盖注册(成功/重复/密码强度 4 类)、密码哈希(加盐随机/可验证)、登录(成功/错误密码/不存在用户)、Refresh Token(成功/无效/类型错误/过期)、受保护接口(无 token/有效/无效)、RBAC、安全头、数据库持久化、日志脱敏共 9 大类 24 个用例;每个用例独立重建内存数据库,互不干扰;mask_sensitive 函数可单独测试,确保日志不会泄露敏感信息。

    10. 总结与进阶方向

    10.1 学习成果回顾

    通过本文的学习,你已经掌握了从零搭建一个 API 的完整流程。下面按「开发阶段」梳理核心知识点:

    阶段核心内容关键技术
    概念理解 API 的本质与组成 端点、方法、参数、响应、状态码
    环境准备 技术选型与项目初始化 Python、FastAPI、Uvicorn
    数据建模 定义数据结构与校验规则 Pydantic、类型提示
    接口开发 实现完整 CRUD 功能 路由、依赖注入、异常处理
    数据持久化 从内存存储升级到数据库 SQLAlchemy、SQLite
    测试调试 验证接口正确性与健壮性 Swagger、curl、自动化脚本
    部署上线 容器化与云平台发布 Docker、云服务
    运维监控 保障服务稳定运行 健康检查、日志、告警

    10.2 核心能力清单

    完成本教程后,你应该具备以下能力:

  • 理解 API 的基本概念:端点、方法、参数、响应、状态码
  • 做好准备工作:技术选型、环境搭建、需求分析
  • 设计数据模型:使用 Pydantic 定义数据结构
  • 实现 CRUD 接口:创建、查询、更新、删除
  • 错误处理:使用全局异常处理器统一返回错误信息
  • 数据库集成:使用 SQLAlchemy 连接 SQLite 实现数据持久化
  • 测试与调试:使用 Swagger、curl 和自动化脚本验证接口
  • 部署上线:使用 Docker 或云平台发布服务
  • 部署后监控:健康检查、结构化日志、监控平台告警
  • 10.3 进阶学习方向

    当你掌握了基础流程后,可以从以下几个方向继续深入。每个方向都给出了具体的学习路径和实战建议:

    10.3.1 数据库与 ORM 进阶
    • 学习目标:从 SQLite 迁移到生产级数据库,掌握迁移工具
    • 具体内容:
      • 使用 PostgreSQL 或 MySQL 替换 SQLite,理解连接池配置
      • 学习 Alembic 数据库迁移工具,管理表结构变更
      • 掌握 SQLAlchemy 2.0 异步 ORM,提升高并发下的性能
    • 实战建议:将待办事项 API 迁移到 PostgreSQL,并编写一个 Alembic 迁移脚本
    10.3.2 认证与授权深化
    • 学习目标:从 JWT 基础认证升级到完整的 OAuth2 授权体系
    • 具体内容:
      • 实现 OAuth2 授权码模式,支持第三方登录(GitHub、Google)
      • 学习 Refresh Token 轮换机制,提升安全性
      • 掌握 Scope 权限粒度控制,实现细粒度授权
    • 实战建议:为 API 添加 GitHub OAuth 登录,并实现基于 Scope 的权限控制
    10.3.3 性能优化与缓存
    • 学习目标:掌握缓存策略和异步编程,提升接口响应速度
    • 具体内容:
      • 使用 Redis 缓存热点数据,减少数据库查询
      • 学习异步数据库驱动(asyncpg、aiomysql)和异步路由
      • 掌握数据库索引优化和查询性能分析(EXPLAIN)
    • 实战建议:为列表接口添加 Redis 缓存,并对比缓存前后的响应耗时
    10.3.4 微服务与消息队列
    • 学习目标:理解微服务架构,掌握服务间通信方式
    • 具体内容:
      • 学习 Docker Compose 编排多服务,理解服务发现
      • 使用 RabbitMQ 或 Kafka 实现异步消息通信
      • 掌握 API 网关(Kong、Traefik)的限流、路由和鉴权
    • 实战建议:将待办事项 API 拆分为用户服务和待办服务,通过消息队列实现异步通知
    10.3.5 可观测性与 DevOps
    • 学习目标:构建完整的监控体系,掌握自动化运维
    • 具体内容:
      • 使用 Prometheus + Grafana 搭建监控看板
      • 学习 OpenTelemetry 实现分布式链路追踪
      • 掌握 Kubernetes 部署和 Helm 包管理
    • 实战建议:为 API 接入 Prometheus 指标采集,并在 Grafana 中创建请求量和错误率看板

    进阶路线建议:建议按「数据库 → 认证 → 性能 → 微服务 → DevOps」的顺序逐步深入,每个方向先完成一个最小实战项目,再进入下一个方向,避免贪多嚼不烂。

    10.4 实践建议

    API 开发是一个不断实践和积累的过程,这里给你几点建议:

  • 动手实践:跟着本文的示例敲一遍代码,不要直接复制粘贴
  • 扩展功能:尝试为待办事项添加分类、优先级、截止日期等字3. 阅读官方文档:深入阅读 FastAPI 和 SQLAlchemy 的官方文档,是提升技能最直接的途径:
    • FastAPI 官方文档:https://fastapi.tiangolo.com/zh/(中文版),涵盖教程、进阶用法、部署与安全最佳实践
    • SQLAlchemy 官方文档:https://docs.sqlalchemy.org/(英文),包含 ORM、Core、方言与迁移等完整参考
    • FastAPI 源码仓库:https://github.com/fastapi/fastapi,可查看最新特性与 issue 讨论
    • SQLAlchemy 源码仓库:https://github.com/sqlalchemy/sqlalchemy,了解底层实现与设计思路文4. 参与开源:在 GitHub 上阅读优秀的 API 项目,学习最佳实践。这里推荐几个高质量的开源项目,涵盖不同技术栈与设计风格,非常适合对照学习:
    • FastAPI 官方示例仓库:https://github.com/fastapi/full-stack-fastapi-template,一个完整的全栈项目模板,包含用户认证、数据库、前端与部署配置,是学习 FastAPI 工程化落地的绝佳范本
    • RealWorld 项目:https://github.com/gothinkster/realworld,用同一套需求规范实现了 50+ 种技术栈的 API 后端,方便横向对比不同框架的代码组织与最佳实践
    • GitHub 官方 REST API 文档:https://docs.github.com/zh/rest,真实世界超大规模 API 的设计范例,可学习版本管理、分页、限流、错误码等生产级设计
    • Stripe API 参考:https://docs.stripe.com/api,业界公认的 API 设计标杆,其资源命名、错误处理与版本策略值得反复研读
    • Public APIs 集合:https://github.com/public-apis/public-apis,收录了上千个免费公开 API,适合用来练习调用与联调实践
  • 持续迭代:将学到的知识应用到实际项目中,不断优化和改进
  • 希望你能在动手实践中不断成长,早日成为一名优秀的后端开发者。祝你编码愉快!

    赞(0)
    未经允许不得转载:171主机测评 » 从 0 到 1 搭建 API:FastAPI 实战教程(含数据库、部署与安全)
    分享到: 更多 (0)

    评论 抢沙发

    • 昵称 (必填)
    • 邮箱 (必填)
    • 网址