FastAPI学习,阶段 1:FastAPI 核心(初识框架)
⚠️ 重要:本文包含 API创建 、路径操作(Path Operations)、查询参数与请求体、响应模型与状态码、错误处理(全局异常捕获与统一错误格式)。
文中涉及的相关代码示例地址:https://github.com/m12305/hello-FastAPI
1.1 Hello World — 第一个 FastAPI 应用
从这里开始,你正式进入 FastAPI 的世界。3 行代码启动一个 API,然后逐层揭开它的面纱。
1. 环境准备
# 确认 Python 版本 >= 3.8
python --version
# 创建虚拟环境(如果还没创建)
python -m venv venv
# 激活虚拟环境
# Windows:
venv\Scripts\activate
# Mac/Linux:
source venv/bin/activate
# 安装 FastAPI 和服务器
pip install fastapi uvicorn[standard]
验证安装:
pip show fastapi # 应该显示版本号,如 0.115.x
pip show uvicorn # 应该显示版本号
2. 最小应用:3 行代码
创建文件并运行你的第一个 FastAPI 应用:
# 文件名: main.py
from fastapi import FastAPI
app = FastAPI() # 创建应用实例
@app.get("/") # 注册路由:GET /
async def root(): # 路径操作函数
return {"message": "Hello World"} # 自动转为 JSON
运行
uvicorn main:app --reload
# │ │ │
# │ │ └── --reload: 代码修改后自动重启(开发模式)
# │ └──────── 应用实例名(变量名)
# └───────────── 文件名(不含 .py)
打开浏览器访问:
- API 地址:http://127.0.0.1:8000 — 看到
{"message":"Hello World"} - 交互式文档:http://127.0.0.1:8000/docs — Swagger UI
- 备选文档:http://127.0.0.1:8000/redoc — ReDoc
🎉 你已经有了一个生产就绪的 API(自动文档、自动 JSON 序列化、高性能异步服务器)。
3. 发生了什么?逐行解析
from fastapi import FastAPI # 导入 FastAPI 类
app = FastAPI() # 实例化一个 FastAPI 应用
# app 是整个应用的核心,所有路由都注册在它上面
# 参数说明(常用):
# title="My API" → 文档标题
# description="..." → 文档描述
# version="1.0.0" → 版本号
# docs_url=None → 禁用 Swagger
# redoc_url=None → 禁用 ReDoc
@app.get("/") # 路径操作装饰器
# @app.get → 处理 GET 请求
# @app.post → 处理 POST 请求
# @app.put → 处理 PUT 请求
# @app.delete → 处理 DELETE 请求
# "/" → URL 路径
async def root(): # 路径操作函数(Path Operation Function)
return {"message": "..."} # 返回 dict → FastAPI 自动转为 JSON
FastAPI 应用初始化常用参数
app = FastAPI(
title="我的第一个 API",
description="这是一个学习 FastAPI 的示例项目",
version="0.1.0",
# docs_url="/api-docs", # 自定义文档路径
# openapi_url="/openapi.json", # 自定义 OpenAPI Schema 路径
)
4. Uvicorn 是什么?
浏览器/客户端 ──HTTP──> Uvicorn (ASGI 服务器) ──ASGI──> FastAPI 应用
<── 响应 <── 返回数据
| 组件 | 角色 |
|---|---|
| Uvicorn | ASGI 服务器 — 接收 HTTP 请求,管理进程/线程,转发给应用 |
| FastAPI | ASGI 应用 — 处理请求(路由、校验、序列化),返回数据 |
| ASGI | 异步服务器网关接口 — Python 异步 Web 服务器和应用之间的协议 |
FastAPI 不包含服务器,它需要 Uvicorn(或 Hypercorn)来接收网络请求。类比:Flask/Django 是 WSGI 应用 + Gunicorn。
5. --reload 开发模式
# 开发模式(代码修改自动重启)
uvicorn main:app --reload
# 等价于
uvicorn main:app --reload --reload-dir ./
# 生产模式(不要用 --reload!性能更好)
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
| 参数 | 说明 | 默认值 |
|---|---|---|
--host |
绑定的 IP | 127.0.0.1 |
--port |
绑定的端口 | 8000 |
--reload |
自动重启 | 关闭 |
--workers |
工作进程数 | 1 |
6. 探索自动生成的文档
启动应用后,访问 http://127.0.0.1:8000/docs
你会看到:
- 📋 你的路由
GET /列在页面上 - 🔽 展开后可以看到 Try it out 按钮
- 点击后可以直接在浏览器中测试 API
- 下面有 Response,展示了返回的 JSON
- 整个页面遵循 OpenAPI 标准
这套文档是从你的代码自动推断的——你的函数参数类型、Pydantic 模型、docstring 都会自动反映在文档中。
7. 测试你的 API
方式 1:浏览器
直接访问 http://127.0.0.1:8000
方式 2:Swagger UI
访问 http://127.0.0.1:8000/docs → Try it out → Execute
方式 3:curl
curl http://127.0.0.1:8000/
# 输出: {"message":"Hello World"}
# 查看响应头
curl -i http://127.0.0.1:8000/
# HTTP/1.1 200 OK
# content-type: application/json
# ...
方式 4:Python httpx
import httpx
resp = httpx.get("http://127.0.0.1:8000/")
print(resp.status_code) # 200
print(resp.json()) # {'message': 'Hello World'}
方式 5:FastAPI TestClient(后面详细学)
from fastapi.testclient import TestClient
client = TestClient(app)
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello World"}
8. 扩展你的第一个 API
# main.py
from fastapi import FastAPI
app = FastAPI(
title="我的第一个 API",
description="FastAPI 学习项目",
version="0.1.0",
)
@app.get("/")
async def root():
"""根路径,返回欢迎信息"""
return {"message": "Hello World", "version": "0.1.0"}
@app.get("/ping")
async def ping():
"""健康检查接口"""
return {"status": "ok", "service": "my-api"}
@app.get("/time")
async def current_time():
"""返回服务器当前时间"""
from datetime import datetime
return {"current_time": datetime.now().isoformat()}
@app.get("/hello/{name}")
async def hello(name: str):
"""向指定用户问好"""
return {"message": f"Hello, {name}!"}
保存后刷新 http://127.0.0.1:8000/docs,你会看到 4 个接口出现在文档中。
9. 常见问题排查
问题 1:端口被占用
# Error: [Errno 10048] address already in use
# 换一个端口
uvicorn main:app --reload --port 8001
问题 2:找不到模块
# ModuleNotFoundError: No module named 'main'
# 确认:
# 1. 你在正确的目录下运行命令
# 2. 文件名是 main.py
# 3. 文件名不含特殊字符
问题 3:没有自动重载
# 确保加了 --reload 参数
# 某些 IDE 保存时不会触发文件变更事件,手动 touch 一下
10. 实战练习
"""练习:创建一个个人简介 API
要求:
1. FastAPI 应用标题为 "个人简介 API"
2. 实现以下接口:
- GET / → 返回 {"name": "你的名字", "role": "FastAPI Learner"}
- GET /about → 返回你的学习目标和开始日期
- GET /status → 返回 {"status": "online", "uptime": "计算从程序启动到现在的秒数"}
提示:uptime 可以在应用启动时记录 start_time = time.time()
"""
# 你的代码:
11. 本章检查清单
- 成功安装了 fastapi 和 uvicorn
- 创建并运行了第一个 FastAPI 应用
- 在浏览器中看到了
{"message": "Hello World"} - 访问了
/docs页面,并试用 Swagger UI - 访问了
/redoc页面 - 理解了
uvicorn main:app --reload每个部分的含义 - 理解了 FastAPI(应用)和 Uvicorn(服务器)的关系
- 用 curl 或 httpx 测试了 API
- 为应用添加了标题和描述
- 完成了实战练习
🎯 恭喜!你已经有了一个可以运行的 API。接下来学路径操作,让 API 拥有真正的 CRUD 能力。
1.2 路径操作(Path Operations)
路径操作是 API 的骨架——URL 设计和 HTTP 方法的组合。掌握它就掌握了路由的完整控制权。
1. 什么是"路径操作"?
@app.get("/users/{user_id}")
\________/ \______________/
│ │
│ └── 路径(Path):URL 路径,可以包含动态参数
└───────────────── 操作(Operation):HTTP 方法(GET/POST/PUT/DELETE/PATCH)
FastAPI 的路径操作 = HTTP 方法装饰器 + URL 路径 + 路径操作函数。
2. 所有 HTTP 方法装饰器
from fastapi import FastAPI
app = FastAPI()
@app.get("/items") # 读取资源列表
async def list_items():
return [{"id": 1}, {"id": 2}]
@app.get("/items/{item_id}") # 读取单个资源(注意:有路径参数)
async def get_item(item_id: int):
return {"id": item_id}
@app.post("/items") # 创建资源
async def create_item():
return {"message": "Item created"}
@app.put("/items/{item_id}") # 完整替换资源
async def replace_item(item_id: int):
return {"message": f"Item {item_id} replaced"}
@app.patch("/items/{item_id}") # 部分更新资源
async def update_item(item_id: int):
return {"message": f"Item {item_id} updated"}
@app.delete("/items/{item_id}") # 删除资源
async def delete_item(item_id: int):
return {"message": f"Item {item_id} deleted"}
# 高级:自定义 HTTP 方法(或同时接受多个方法)
@app.api_route("/custom", methods=["GET", "POST", "PURGE"])
async def custom_handler():
return {"message": "custom"}
3. 路径参数(Path Parameters)
路径参数是 URL 中动态的部分,用 {} 包裹:
@app.get("/users/{user_id}")
async def get_user(user_id: int): # ← 函数参数名必须与 {user_id} 一致
return {"user_id": user_id}
# GET /users/42 → {"user_id": 42}
# GET /users/abc → 自动返回 422 错误(类型不匹配,期望 int)
类型转换
FastAPI 根据函数参数的类型标注自动转换:
@app.get("/items/{item_id}")
async def get_item(item_id: int): # ← 自动转为 int
return {"item_id": item_id, "type": type(item_id).__name__}
# GET /items/123 → {"item_id": 123, "type": "int"}
@app.get("/price/{amount}")
async def get_price(amount: float): # ← 自动转为 float
...
# GET /price/19.99 → amount = 19.99
@app.get("/flag/{value}")
async def get_flag(value: bool): # ← 自动转为 bool
...
# GET /flag/true → value = True
# GET /flag/1 → value = True
# GET /flag/false → value = False
4. 路径参数的顺序与路由优先级
# ⚠️ 重要:固定路径必须写在动态路径前面!
# ✅ 正确顺序
@app.get("/users/me") # 先注册固定路径
async def get_current_user():
return {"username": "Alice"}
@app.get("/users/{user_id}") # 再注册动态路径
async def get_user(user_id: int):
return {"user_id": user_id}
# ❌ 错误顺序:/users/{user_id} 会吞掉 /users/me
# GET /users/me → user_id="me" → 类型转换 int("me") → 报错
5. 路径参数的枚举约束
当路径参数的取值只有几种可能时,用枚举限制:
from enum import Enum
class Category(str, Enum):
electronics = "electronics"
clothing = "clothing"
food = "food"
@app.get("/products/{category}")
async def get_products(category: Category):
"""category 只能是 electronics, clothing, food"""
return {"category": category, "products": [...]}
# GET /products/electronics → ✅
# GET /products/books → ❌ 422 错误,自动提示可选值
枚举的好处:
- 类型安全 — 非法值自动拒绝
- 文档自描述 — Swagger 中自动显示可选值作为下拉菜单
6. 定义多个路径参数的场景
# 层级嵌套:一个用户的多篇文章
@app.get("/users/{user_id}/posts/{post_id}")
async def get_user_post(user_id: int, post_id: int):
return {"user_id": user_id, "post_id": post_id}
# GET /users/42/posts/5 → {"user_id": 42, "post_id": 5}
# 更复杂的层级
@app.get("/orgs/{org_id}/teams/{team_id}/members/{member_id}")
async def get_team_member(org_id: int, team_id: int, member_id: int):
return {
"org": org_id,
"team": team_id,
"member": member_id,
}
7. @app.api_route() 高级用法
当你需要一个路由同时响应多种 HTTP 方法时:
@app.api_route("/webhook", methods=["GET", "POST"])
async def webhook():
"""同一个处理函数同时处理 GET 和 POST"""
return {"status": "received"}
# 或者对请求对象做精细控制
from fastapi import Request
@app.api_route("/dynamic", methods=["GET", "POST", "PUT", "DELETE"])
async def dynamic_handler(request: Request):
"""根据 request.method 做不同处理"""
method = request.method
if method == "GET":
return {"action": "read"}
elif method == "POST":
return {"action": "create"}
elif method == "PUT":
return {"action": "replace"}
else:
return {"action": "delete"}
8. 路径操作装饰器的额外参数
@app.get(
"/items/{item_id}",
response_model=ItemResponse, # 响应模型(第 1.4 章详讲)
status_code=200, # 默认成功状态码
tags=["items"], # Swagger 文档中的分组标签
summary="获取单个商品", # Swagger 中的简短摘要
description="通过商品 ID 获取商品的详细信息", # 详细描述
deprecated=True, # 标记为已废弃(文档中会显示)
)
async def get_item(item_id: int):
...
这些参数影响的是文档展示和响应行为,不影响路由匹配。
9. 实战:用临时字典模拟数据库
from fastapi import FastAPI
from enum import Enum
app = FastAPI(title="TODO API")
# 用字典当临时数据库
todos: dict[int, dict] = {}
next_id = 1
@app.get("/todos")
async def list_todos():
"""获取所有 Todo"""
return list(todos.values())
@app.get("/todos/{todo_id}")
async def get_todo(todo_id: int):
"""获取单个 Todo"""
if todo_id not in todos:
from Fastapi import HTTPException
raise HTTPException(status_code=404, detail="Todo not found")
return todos[todo_id]
@app.post("/todos")
async def create_todo(title: str, done: bool = False):
"""创建新的 Todo"""
global next_id
todos[next_id] = {"id": next_id, "title": title, "done": done}
next_id += 1
@app.put("/todos/{todo_id}")
async def update_todo(todo_id: int, title: str, done: bool):
"""完整更新 Todo"""
if todo_id not in todos:
from Fastapi import HTTPException
raise HTTPException(status_code=404, detail="Todo not found")
...
@app.delete("/todos/{todo_id}")
async def delete_todo(todo_id: int):
"""删除 Todo"""
if todo_id not in todos:
from Fastapi import HTTPException
raise HTTPException(status_code=404, detail="Todo not found")
...
注意: 上面用函数参数
title: str直接接收数据,这只是临时写法。下一章(1.3)会学到正确的请求体处理方式。
10. 实践练习
"""练习:书籍管理 API 骨架
要求:
1. 定义 BookCategory 枚举(fiction, non-fiction, science, history)
2. 实现以下路由:
- GET /books → 返回书籍列表(用硬编码列表)
- GET /books/{book_id} → 返回单本书(不存在返回 404)
- GET /books/category/{category} → 按分类筛选
- POST /books → 创建书籍(用临时字典存储)
- DELETE /books/{book_id} → 删除书籍
3. 给每个路由添加 tags、summary 参数
4. 访问 /docs 确认文档展示正常
"""
# 你的代码:
11. 本章检查清单
- 理解路径操作 = HTTP 方法 + URL 路径 + 处理函数
- 能使用全部 6 种装饰器(get, post, put, patch, delete, api_route)
- 理解路径参数
{param}的语法和类型自动转换 - 知道固定路径要写在动态路径前面
- 能用枚举约束路径参数的取值
- 能设计嵌套的 URL 层级结构
- 了解装饰器的 tags、summary、deprecated 等文档参数
- 完成实战练习
🎯 路径操作是 API 的入口。下一章学习查询参数和请求体,让 API 能真正处理用户提交的数据。
1.3 查询参数与请求体
API 的核心价值是接收和处理数据。这一章你将学会接收查询参数和 JSON 请求体的正确姿势。
1. 三种数据来源总览
POST /users/42?lang=zh¬ify=true
\________/ \___________/ \____________________/
│ │ │
请求体 路径参数 查询参数
(JSON) (URL 一部分) (?key=value)
| 数据来源 | 在 URL 中 | 必选/可选 | 典型用途 |
|---|---|---|---|
| 路径参数 | /users/{id} |
必选 | 资源唯一标识 |
| 查询参数 | ?page=1&size=20 |
可选(有默认值) | 过滤、排序、分页 |
| 请求体 | HTTP Body(JSON) | 看 API 设计 | 创建/更新时的数据 |
2. 查询参数(Query Parameters)
基础用法
from fastapi import FastAPI
app = FastAPI()
# 函数参数中不在路径中的 → 自动成为查询参数
@app.get("/users")
async def list_users(
page: int = 1, # 有默认值 → 可选参数,?page=2
size: int = 20, # 有默认值 → 可选参数,?size=50
active_only: bool = False,
):
return {
"page": page,
"size": size,
"active_only": active_only,
}
# GET /users → 使用所有默认值
# GET /users?page=2 → page=2, 其余默认
# GET /users?page=2&size=50&active_only=true → 全部指定
必选 vs 可选
@app.get("/search")
async def search(
q: str, # ← 没有默认值 → 必选!
limit: int = 10, # ← 有默认值 → 可选
):
return {"q": q, "limit": limit}
# GET /search → 422 错误(缺少必选参数 q)
# GET /search?q=python → ✅ {"q": "python", "limit": 10}
可选参数用 Optional
from typing import Optional
@app.get("/users")
async def list_users(
status: Optional[str] = None, # 可以为 None,不传就是 None
tag: str | None = None, # Python 3.10+ 写法
):
filters = {}
if status:
filters["status"] = status
if tag:
filters["tag"] = tag
return {"filters": filters}
3. 用 Query() 添加高级校验
当需要给查询参数添加校验规则和元数据时,用 Query():
from fastapi import FastAPI, Query
from typing import Annotated
app = FastAPI()
@app.get("/items")
async def list_items(
# Annotated 写法(FastAPI 推荐)
page: Annotated[int, Query(ge=1, description="页码")] = 1,
size: Annotated[int, Query(ge=1, le=100, description="每页数量")] = 20,
# 等价传统写法(不推荐,类型标注和校验混在一起)
# page: int = Query(1, ge=1, description="页码"),
):
return {"page": page, "size": size}
# GET /items?page=0 → 422 错误(page 必须 >= 1)
# GET /items?size=200 → 422 错误(size 必须 <= 100)
Query() 常用参数
@app.get("/search")
async def search_items(
q: Annotated[
str,
Query(
min_length=1, # 最小长度
max_length=50, # 最大长度
pattern=r"^[a-zA-Z0-9 ]+$", # 正则匹配
title="搜索关键词",
description="搜索商品名称或描述",
alias="query", # URL 中使用 ?query=xxx 而不是 ?q=xxx
deprecated=True, # 在文档中标记为废弃
include_in_schema=False, # 从 OpenAPI 文档中隐藏
),
] = None,
):
return {"q": q}
4. 请求体(Request Body)
请求体是 POST/PUT/PATCH 时通过 HTTP Body 发送的数据,通常为 JSON 格式。
用 Pydantic 模型定义请求体
from pydantic import BaseModel, Field
class ItemCreate(BaseModel):
name: str = Field(min_length=1, max_length=100)
price: float = Field(gt=0)
description: str | None = Field(default=None, max_length=1000)
tax: float | None = None
@app.post("/items")
async def create_item(item: ItemCreate): # ← Pydantic 模型 = 请求体
# item 已经是校验过的 ItemCreate 实例
return {
"name": item.name,
"price": item.price,
"price_with_tax": item.price + (item.tax or 0),
}
测试请求体
curl -X POST http://127.0.0.1:8000/items \
-H "Content-Type: application/json" \
-d '{"name": "Widget", "price": 19.99, "tax": 1.5}'
或在 Swagger UI (/docs) 中直接填入 JSON。
5. 路径参数 + 查询参数 + 请求体 组合
from pydantic import BaseModel
class OrderCreate(BaseModel):
product_id: int
quantity: int = Field(ge=1)
address: str
@app.post("/users/{user_id}/orders")
async def create_order(
user_id: int, # 路径参数
order: OrderCreate, # 请求体
send_notification: bool = False, # 查询参数
priority: str = "normal", # 查询参数
):
return {
"message": f"Order created for user {user_id}",
"product_id": order.product_id,
"quantity": order.quantity,
"address": order.address,
"notification": send_notification,
"priority": priority,
}
# POST /users/42/orders?send_notification=true&priority=high
# Body: {"product_id": 101, "quantity": 3, "address": "Beijing"}
#
# FastAPI 自动把数据分配到正确的参数:
# - user_id=42 → 来自 URL 路径
# - order → 来自请求体 JSON
# - send_notification → 来自查询参数
# - priority → 来自查询参数
6. 多请求体参数
当一个接口需要接收多个逻辑上独立的数据块时:
class UserInfo(BaseModel):
name: str
email: str
class ShippingAddress(BaseModel):
street: str
city: str
zip_code: str
@app.post("/checkout")
async def checkout(
user: UserInfo, # 第一个 JSON 对象 {"name":..., "email":...}
address: ShippingAddress,# 第二个 JSON 对象 {"street":..., ...}
):
return {
"user": user,
"shipping": address,
}
# 请求体需要这样发送:
# {
# "user": {"name": "Alice", "email": "alice@example.com"},
# "address": {"street": "123 Main St", "city": "Beijing", "zip_code": "100000"}
# }
用 Body(embed=True) 嵌入单个字段
from fastapi import Body
@app.post("/users")
async def create_user(
name: str = Body(...), # ... 表示必填
email: str = Body(...),
age: int = Body(25), # 可选 Body 参数
):
return {"name": name, "email": email, "age": age}
# 外部调用时需要包裹一层:
# {"name": "Alice", "email": "alice@example.com", "age": 25}
7. 数据来源的优先级与识别规则
FastAPI 的参数识别规则(按优先级):
@app.get("/items/{item_id}")
async def some_handler(
item_id: int, # 1. 在路径中 → 路径参数
query_param: str = "xxx", # 2. 基本类型 + 不在路径中 → 查询参数
body_param: MyModel, # 3. Pydantic 模型 → 请求体
special: int = Body(...), # 4. 显式用了 Body() → 请求体
header: str = Header(...), # 5. 显式用了 Header() → 请求头
cookie: str = Cookie(...), # 6. 显式用了 Cookie() → Cookie
):
...
8. 实战:构建一个完整的商品搜索 API
from fastapi import FastAPI, Query, Body
from pydantic import BaseModel, Field
from typing import Annotated, Optional
from enum import Enum
app = FastAPI(title="商品管理 API")
# 模拟数据库
products_db: list[dict] = [
{"id": 1, "name": "机械键盘", "category": "electronics", "price": 399.0, "in_stock": True},
{"id": 2, "name": "Python 编程书", "category": "books", "price": 79.0, "in_stock": True},
{"id": 3, "name": "T恤", "category": "clothing", "price": 129.0, "in_stock": False},
]
class ProductCreate(BaseModel):
name: str = Field(min_length=1, max_length=100)
category: str
price: float = Field(gt=0)
in_stock: bool = True
# GET /products —— 搜索 + 筛选 + 分页
@app.get("/products")
async def search_products(
q: Annotated[Optional[str], Query(description="搜索关键词")] = None,
category: Annotated[Optional[str], Query(description="分类筛选")] = None,
min_price: Annotated[Optional[float], Query(ge=0, description="最低价")] = None,
max_price: Annotated[Optional[float], Query(ge=0, description="最高价")] = None,
page: Annotated[int, Query(ge=1)] = 1,
size: Annotated[int, Query(ge=1, le=50)] = 10,
):
results = products_db
if q:
results = [p for p in results if q.lower() in p["name"].lower()]
if category:
results = [p for p in results if p["category"] == category]
if min_price is not None:
results = [p for p in results if p["price"] >= min_price]
if max_price is not None:
results = [p for p in results if p["price"] <= max_price]
# 分页
start = (page - 1) * size
end = start + size
return {
"total": len(results),
"page": page,
"size": size,
"items": results[start:end],
}
# POST /products —— 创建商品
@app.post("/products", status_code=201)
async def create_product(product: ProductCreate):
new_id = max(p["id"] for p in products_db) + 1
new_product = {"id": new_id, **product.model_dump()}
products_db.append(new_product)
return new_product
9. 实战练习
"""练习:图书管理 API
实现以下接口:
1. POST /books —— 创建图书
请求体包含:title(必填), author(必填), year(必填,1900-2026),
pages(可选,>0), isbn(可选,13位数字)
2. GET /books —— 搜索图书
查询参数:author(可选), min_year(可选), max_year(可选),
keyword(可选,搜索 title),
page(>=1), size(1-100)
要求:
- 使用 Pydantic 模型定义请求体
- 使用 Annotated[..., Query(...)] 添加校验
- 用内存列表存储数据
"""
# 你的代码:
10. 本章检查清单
- 理解三种数据来源:路径参数、查询参数、请求体的区别
- 能区分必选参数和可选参数
- 能用
Query()给查询参数添加校验规则 - 能用 Pydantic 模型定义请求体
- 能在同一个路由中混合使用三种数据来源
- 理解 FastAPI 的参数识别规则
- 能实现带搜索、筛选、分页的列表接口
- 完成实战练习
🎯 数据输入掌握了。下一章学响应模型和状态码,掌控 API 的输出。
1.4 响应模型与状态码
控制 API 的输出和输入同样重要。这一章教你精确控制返回什么数据、以什么格式、带什么状态码。
1. 默认行为
from fastapi import FastAPI
app = FastAPI()
@app.get("/users/{user_id}")
async def get_user(user_id: int):
return {"id": user_id, "name": "Alice", "password_hash": "abc123"}
FastAPI 默认:
- 把你 return 的任何东西序列化为 JSON
- 状态码默认 200(GET/PUT/PATCH)或 201(POST)
- 不限制返回的字段
2. response_model —— 控制响应结构
response_model 是 FastAPI 最重要的响应控制手段:
from pydantic import BaseModel
# 数据库中的用户(包含敏感字段)
fake_db = {
1: {"id": 1, "name": "Alice", "email": "alice@example.com",
"password_hash": "$2b$12$...", "internal_notes": "VIP customer"}
}
# 公开的响应模型(只暴露安全字段)
class UserResponse(BaseModel):
id: int
name: str
email: str
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
user = fake_db.get(user_id)
# user 包含 password_hash 和 internal_notes
return user
# FastAPI 自动过滤:只返回 UserResponse 中定义的字段!
# {"id": 1, "name": "Alice", "email": "alice@example.com"}
为什么用 response_model?
- 安全 — 绝不泄露密码哈希、内部备注等敏感数据
- 文档 — Swagger 自动展示响应结构
- 校验 — 确保返回的数据符合 Schema
- 性能 — 过滤掉不需要返回的大字段
3. 响应字段的过滤
response_model_include / exclude
class UserResponse(BaseModel):
id: int
name: str
email: str
is_active: bool
created_at: str
# 只看部分字段
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user_basic(user_id: int):
...
# 只包含指定字段
@app.get("/users/{user_id}", response_model=UserResponse,
response_model_include={"id", "name"})
async def get_user_minimal(user_id: int):
...
# 返回: {"id": 1, "name": "Alice"}
# 排除指定字段
@app.get("/users/{user_id}", response_model=UserResponse,
response_model_exclude={"created_at", "is_active"})
async def get_user_clean(user_id: int):
...
# 返回: {"id": 1, "name": "Alice", "email": "alice@example.com"}
嵌套列表的响应模型
class ItemResponse(BaseModel):
id: int
name: str
price: float
# 返回列表时,用 List[ItemResponse]
from typing import List
@app.get("/items", response_model=List[ItemResponse])
async def list_items():
return [
{"id": 1, "name": "Widget", "price": 19.99, "_internal": "xxx"},
{"id": 2, "name": "Gadget", "price": 29.99, "_internal": "yyy"},
]
# 每个元素都会被 ItemResponse 过滤
4. 状态码(Status Code)
设置默认状态码
from fastapi import FastAPI, status
@app.post("/users", status_code=201) # 数字
async def create_user():
return {"message": "created"}
@app.post("/users", status_code=status.HTTP_201_CREATED) # 推荐:用 status 常量
async def create_user():
return {"message": "created"}
# 常用 status 常量:
# status.HTTP_200_OK
# status.HTTP_201_CREATED
# status.HTTP_204_NO_CONTENT
# status.HTTP_400_BAD_REQUEST
# status.HTTP_401_UNAUTHORIZED
# status.HTTP_403_FORBIDDEN
# status.HTTP_404_NOT_FOUND
# status.HTTP_422_UNPROCESSABLE_ENTITY
# status.HTTP_500_INTERNAL_SERVER_ERROR
动态设置状态码
from fastapi import Response
@app.get("/items/{item_id}")
async def get_item(item_id: int, response: Response):
if item_id == 99:
response.status_code = 302 # 临时重定向
response.headers["Location"] = "/items/100"
return None
return {"id": item_id}
@app.post("/items")
async def create_item(item: ItemCreate, response: Response):
# 创建成功,手动设置 201 + Location 头
response.status_code = status.HTTP_201_CREATED
response.headers["Location"] = f"/items/{new_id}"
return new_item
5. 响应类型大全
除了默认的 JSON 响应,FastAPI 支持多种响应类型:
from fastapi import FastAPI
from fastapi.responses import (
JSONResponse,
HTMLResponse,
PlainTextResponse,
FileResponse,
StreamingResponse,
RedirectResponse,
)
app = FastAPI()
# 1. JSON 响应(默认,也可以显式使用)
@app.get("/json")
async def json_response():
return JSONResponse(
content={"message": "hello"},
status_code=200,
headers={"X-Custom-Header": "value"},
)
# 2. HTML 响应(配合 Jinja2 模板)
@app.get("/html", response_class=HTMLResponse)
async def html_response():
return """
<html>
<head><title>FastAPI</title></head>
<body><h1>Hello FastAPI!</h1></body>
</html>
"""
# 3. 纯文本
@app.get("/text", response_class=PlainTextResponse)
async def text_response():
return "Hello, this is plain text"
# 4. 文件下载
@app.get("/download")
async def download_file():
return FileResponse(
path="path/to/file.pdf",
filename="报告.pdf", # 下载时的文件名
media_type="application/pdf",
)
# 5. 流式响应(大文件、实时数据)
@app.get("/stream")
async def stream_data():
async def generate():
for i in range(10):
yield f"data: chunk {i}\n\n"
return StreamingResponse(generate(), media_type="text/plain")
# 6. 重定向
@app.get("/old-path")
async def old_path():
return RedirectResponse(url="/new-path", status_code=301)
6. 响应的最佳实践:请求/响应分离
from pydantic import BaseModel, Field
from datetime import datetime
# ===== 输入模型 =====
class UserCreate(BaseModel):
name: str = Field(min_length=2)
email: str
password: str = Field(min_length=8)
# ===== 输出模型 =====
class UserResponse(BaseModel):
id: int
name: str
email: str
created_at: str
# ===== 数据库模型(SQLAlchemy 前先用 dataclass 模拟) =====
class UserInDB(BaseModel):
"""内部模型,包含所有字段"""
id: int
name: str
email: str
password_hash: str
created_at: str = datetime.now().isoformat()
# ===== 路由 =====
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user_in: UserCreate):
"""创建用户:接收 UserCreate,返回 UserResponse(无密码!)"""
# 模拟:创建数据库记录
db_user = UserInDB(
id=1,
name=user_in.name,
email=user_in.email,
password_hash=f"hashed_{user_in.password}",
)
return db_user # response_model 自动过滤掉 password_hash
黄金法则: 永远不要把数据库模型直接返回给客户端。用专门的响应模型过滤敏感字段。
7. response_model 与类型安全
response_model 还会在返回时校验数据:
class UserResponse(BaseModel):
id: int
name: str
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
# 假设某天数据库返回了错误类型
return {"id": "not-a-number", "name": "Alice"}
# ❌ FastAPI 会抛异常,因为 id 应该是 int 而不是 str
# 这帮你发现数据层的 bug!
8. Response 对象 vs response_model
| 方式 | 校验 | 过滤 | 文档 | 适用场景 |
|---|---|---|---|---|
response_model=PydanticModel |
✅ | ✅ | ✅ | 99% 的 REST API |
Response(content=...) |
❌ | ❌ | ❌ | 需要完全控制响应时 |
JSONResponse(content=...) |
❌ | ❌ | ❌ | 自定义 JSON 响应头 |
return dict |
❌ | ❌ | ✅ | 快速原型 |
9. 实战练习
"""练习:用户管理 API 的响应设计
1. 定义以下模型:
- UserCreate(输入):name, email, password(>=8位)
- UserResponse(公开输出):id, name, email, is_active, created_at
- UserDetailResponse(详情输出):包含 UserResponse + 最近登录时间
2. 实现接口:
- POST /users → 201, response_model=UserResponse
- GET /users → 200, response_model=List[UserResponse]
- GET /users/{id} → 200, response_model=UserDetailResponse
- DELETE /users/{id} → 204 (No Content)
3. 要求:
- 密码绝不返回
- 创建时返回 Location 头
- 删除成功返回 204 无内容
"""
# 你的代码:
10. 本章检查清单
- 理解
response_model的四大作用:安全、文档、校验、性能 - 会用
response_model_include/exclude过滤字段 - 能用
status_code和status.HTTP_*常量设置状态码 - 了解 6 种常用响应类型及其使用场景
- 理解请求模型和响应模型分离的设计原则
- 知道如何返回列表的响应模型
- 知道
response_model也会校验返回数据 - 完成实战练习
🎯 输入和输出都掌握了。下一章学错误处理,让你的 API 健壮而专业。
1.5 错误处理
一个好的 API 不仅能在正常时工作,更要在出错时给出清晰、专业的反馈。这一章教你掌控所有异常场景。
1. HTTPException —— 主动抛出 HTTP 错误
from fastapi import FastAPI, HTTPException, status
app = FastAPI()
fake_items = {1: {"id": 1, "name": "Widget"}}
@app.get("/items/{item_id}")
async def get_item(item_id: int):
if item_id not in fake_items:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, # 状态码
detail=f"Item {item_id} not found", # 错误描述(必填)
headers={"X-Error-Code": "ITEM_NOT_FOUND"}, # 可选的额外响应头
)
return fake_items[item_id]
# GET /items/999
# → 404
# → {"detail": "Item 999 not found"}
# → 响应头: x-error-code: ITEM_NOT_FOUND
HTTPException 参数
| 参数 | 类型 | 说明 |
|---|---|---|
status_code |
int | HTTP 状态码 |
detail |
str / dict / list | 错误详情(可以是 JSON 可序列化的任意类型) |
headers |
dict | 额外的响应头 |
2. 不同的错误,不同的状态码
fake_users = {1: {"id": 1, "name": "Alice", "role": "user"}}
@app.get("/users/{user_id}")
async def get_user(user_id: int):
# 404 — 资源不存在
if user_id not in fake_users:
raise HTTPException(status_code=404, detail="用户不存在")
return fake_users[user_id]
@app.patch("/users/{user_id}/role")
async def update_role(user_id: int, new_role: str):
# 404
if user_id not in fake_users:
raise HTTPException(status_code=404, detail="用户不存在")
# 422 — 参数语义错误
allowed_roles = {"admin", "editor", "viewer"}
if new_role not in allowed_roles:
raise HTTPException(
status_code=422,
detail=f"无效的角色:{new_role}。可选:{allowed_roles}"
)
fake_users[user_id]["role"] = new_role
return fake_users[user_id]
@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
# 409 — 冲突
if fake_users[user_id]["role"] == "admin":
raise HTTPException(status_code=409, detail="不能删除管理员用户")
del fake_users[user_id]
return {"message": "删除成功"}
场景 → 状态码速查
| 场景 | 状态码 |
|---|---|
| 资源不存在 | 404 Not Found |
| 参数校验失败 | 422 Unprocessable Entity |
| 未登录/Token 无效 | 401 Unauthorized |
| 已登录但权限不足 | 403 Forbidden |
| 业务冲突(重复操作等) | 409 Conflict |
| 请求过于频繁 | 429 Too Many Requests |
| 服务不可用 | 503 Service Unavailable |
3. 自定义异常处理器
你可以全局接管特定异常的处理方式:
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
app = FastAPI()
# ===== 1. 自定义异常类 =====
class BusinessError(Exception):
def __init__(self, message: str, error_code: str = "BUSINESS_ERROR"):
self.message = message
self.error_code = error_code
class ItemNotFoundError(BusinessError):
def __init__(self, item_id: int):
super().__init__(
message=f"商品 {item_id} 不存在",
error_code="ITEM_NOT_FOUND"
)
# ===== 2. 注册全局异常处理器 =====
@app.exception_handler(BusinessError)
async def business_error_handler(request: Request, exc: BusinessError):
return JSONResponse(
status_code=400,
content={
"error": exc.error_code,
"message": exc.message,
},
)
# ===== 3. 在路由中使用 =====
@app.get("/items/{item_id}")
async def get_item(item_id: int):
if item_id not in fake_items:
raise ItemNotFoundError(item_id) # 自动被上方处理器捕获
return fake_items[item_id]
4. 覆盖 FastAPI 内置的异常处理器
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
"""自定义参数校验错误的返回格式"""
# 提取所有校验错误
errors = []
for error in exc.errors():
errors.append({
"field": " -> ".join(str(loc) for loc in error["loc"]),
"message": error["msg"],
"type": error["type"],
})
return JSONResponse(
status_code=422,
content={
"error": "VALIDATION_ERROR",
"message": "请求参数校验失败",
"details": errors,
},
)
# 现在所有参数校验错误都会返回这种格式,而不是默认格式
默认格式 vs 自定义格式
// FastAPI 默认格式:
{"detail": [{"loc": ["body", "name"], "msg": "field required", "type": "missing"}]}
// 自定义后:
{
"error": "VALIDATION_ERROR",
"message": "请求参数校验失败",
"details": [
{"field": "body -> name", "message": "field required", "type": "missing"}
]
}
5. 设计统一的错误响应格式
企业级 API 通常有统一的错误响应结构:
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
import traceback
app = FastAPI()
# ===== 统一错误响应模型 =====
def error_response(
status_code: int,
error_type: str,
message: str,
details: list | None = None,
request_id: str | None = None,
) -> dict:
return {
"success": False,
"error": {
"type": error_type,
"code": status_code,
"message": message,
"details": details or [],
"request_id": request_id,
},
}
# ===== 处理 HTTPException =====
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
return JSONResponse(
status_code=exc.status_code,
content=error_response(
status_code=exc.status_code,
error_type="HTTP_ERROR",
message=str(exc.detail),
),
)
# ===== 处理参数校验错误 =====
@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
details = [
{"field": " -> ".join(str(l) for l in e["loc"]), "message": e["msg"]}
for e in exc.errors()
]
return JSONResponse(
status_code=422,
content=error_response(
status_code=422,
error_type="VALIDATION_ERROR",
message="请求参数校验失败",
details=details,
),
)
# ===== 兜底:捕获所有未处理异常 =====
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
# 生产环境中,这里应该记录日志、发送告警
print("UNHANDLED ERROR:", traceback.format_exc())
return JSONResponse(
status_code=500,
content=error_response(
status_code=500,
error_type="INTERNAL_ERROR",
message="服务器内部错误,请稍后重试",
),
)
6. 错误的 HTTPException 使用方式
# ✅ 好:清晰的错误信息
raise HTTPException(
status_code=404,
detail="Task with id 42 not found"
)
# ✅ 好:detail 可以是 dict(给前端更多信息)
raise HTTPException(
status_code=409,
detail={
"message": "用户名已存在",
"suggestion": "请更换用户名或尝试登录",
"field": "username",
}
)
# ❌ 坏:信息太模糊
raise HTTPException(status_code=400, detail="Error")
# ❌ 坏:泄露内部信息
raise HTTPException(status_code=500, detail=traceback.format_exc())
# ✅ 好:生产环境只返回通用信息,内部错误记录到日志
raise HTTPException(status_code=500, detail="服务器内部错误")
7. 实战:完整的错误处理方案
from fastapi import FastAPI, HTTPException, Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from pydantic import BaseModel, Field, field_validator
app = FastAPI(title="错误处理示例")
# ---------- 数据 ----------
users_db = {
1: {"id": 1, "username": "alice", "email": "alice@example.com", "is_active": True},
}
# ---------- 模型 ----------
class UserCreate(BaseModel):
username: str = Field(min_length=3, max_length=20)
email: str
age: int = Field(ge=18, le=120)
@field_validator("username")
@classmethod
def username_no_spaces(cls, v: str) -> str:
if " " in v:
raise ValueError("用户名不能包含空格")
return v
# ---------- 全局异常处理器 ----------
@app.exception_handler(HTTPException)
async def custom_http_handler(request: Request, exc: HTTPException):
return JSONResponse(
status_code=exc.status_code,
content={
"success": False,
"error": str(exc.detail),
"status_code": exc.status_code,
},
)
@app.exception_handler(Exception)
async def global_handler(request: Request, exc: Exception):
return JSONResponse(
status_code=500,
content={
"success": False,
"error": "服务器内部错误",
"status_code": 500,
},
)
# ---------- 路由 ----------
@app.get("/users/{user_id}")
async def get_user(user_id: int):
"""获取用户:404 示例"""
if user_id not in users_db:
raise HTTPException(
status_code=404,
detail=f"用户 {user_id} 不存在"
)
return {"success": True, "data": users_db[user_id]}
@app.post("/users", status_code=201)
async def create_user(user: UserCreate):
"""创建用户:409 冲突示例"""
# 检查用户名重复
for u in users_db.values():
if u["username"] == user.username:
raise HTTPException(
status_code=409,
detail={"message": "用户名已存在", "field": "username"}
)
new_id = max(users_db) + 1
users_db[new_id] = {"id": new_id, **user.model_dump(), "is_active": True}
return {"success": True, "data": users_db[new_id]}
@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
"""删除用户:403 示例"""
if user_id not in users_db:
raise HTTPException(status_code=404, detail=f"用户 {user_id} 不存在")
if user_id == 1:
raise HTTPException(status_code=403, detail="不能删除超级管理员")
del users_db[user_id]
return {"success": True, "message": "删除成功"}
8. 实战练习
"""练习:库存管理 API 的错误处理
实现以下接口,每个接口都要有对应的错误处理:
1. GET /products/{product_id}
- 不存在 → 404
2. POST /products
- name 和 price 必填(依赖 Pydantic 自动校验)
- name 重复 → 409
3. POST /orders
- product_id 不存在 → 404
- 库存不足 → 409(附上当前库存量)
- quantity <= 0 → 422
4. 添加一个全局异常处理器,统一错误响应格式为:
{"error": true, "code": 状态码, "message": "错误信息"}
5. 添加一个自定义的 InsufficientStockError 异常类和对应的处理器
"""
# 你的代码:
9. 本章检查清单
- 能用
HTTPException主动抛出 HTTP 错误 - 能为不同场景选择正确的 HTTP 状态码
- 能自定义异常处理器
@app.exception_handler() - 能覆盖 FastAPI 内置的校验错误处理器
- 能为 API 设计统一的错误响应格式
- 理解生产环境中不要泄露内部错误信息的原则
- 能用
detail传递结构化的错误信息 - 完成实战练习
10. 阶段 1 总结
你已经学完了 FastAPI 核心的 5 个章节:
| 章节 | 核心能力 |
|---|---|
| 1.1 Hello World | 创建应用、运行服务器、访问文档 |
| 1.2 路径操作 | 定义路由、路径参数、HTTP 方法 |
| 1.3 查询参数与请求体 | 接收数据、Pydantic 校验、Query() 参数 |
| 1.4 响应模型与状态码 | 控制输出、过滤敏感字段、设置状态码 |
| 1.5 错误处理 | 抛出异常、统一错误格式、全局异常处理器 |
你现在可以: 构建一个完整的、带数据校验、有专业错误处理的 CRUD API。
🎯 阶段 1 完成! 回顾所有章节的检查清单,确认全部勾选后,进入阶段 2:请求与响应,学习依赖注入、中间件、文件上传等进阶主题。
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。
更多推荐



所有评论(0)