响应模型
响应模型定义了 API 返回数据的结构和格式。通过声明响应模型,FastAPI 能够自动验证响应数据、生成文档,并过滤敏感信息。
返回类型注解
最简单的方式是直接在函数上添加返回类型注解:
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
"""商品模型"""
name: str
price: float
description: str | None = None
@app.post("/items/")
async def create_item(item: Item) -> Item:
return item
@app.get("/items/")
async def read_items() -> list[Item]:
return [
Item(name="商品1", price=99.9),
Item(name="商品2", price=199.9),
]
FastAPI 会使用这个返回类型来:
- 验证返回的数据是否符合定义
- 自动将数据转换为正确的 JSON 格式
- 在 OpenAPI 文档中生成响应模型
response_model 参数
当返回类型与实际需要返回的数据类型不同时,使用 response_model 参数:
from typing import Any
@app.post("/items/", response_model=Item)
async def create_item(item: Item) -> Any:
# 可以返回字典,FastAPI 会根据 response_model 转换
return {"name": "商品名称", "price": 99.9, "extra": "会被过滤"}
response_model 与返回类型注解的区别:
| 特性 | 返回类型注解 | response_model |
|---|---|---|
| 编辑器支持 | 提供类型检查和自动补全 | 不影响函数内部类型检查 |
| 数据过滤 | 会过滤 | 会过滤 |
| 优先级 | 较低 | 较高(同时存在时优先使用) |
| 使用场景 | 返回数据与声明一致时 | 返回数据类型不同时 |
数据过滤
响应模型最重要的功能之一是过滤敏感数据。
输入输出模型分离
创建不同的模型用于输入和输出,避免泄露敏感信息:
from pydantic import BaseModel, EmailStr
class UserIn(BaseModel):
"""用户输入模型(包含密码)"""
username: str
password: str
email: EmailStr
full_name: str | None = None
class UserOut(BaseModel):
"""用户输出模型(不包含密码)"""
username: str
email: EmailStr
full_name: str | None = None
@app.post("/users/", response_model=UserOut)
async def create_user(user: UserIn):
# 即便返回了包含密码的 user 对象
# 响应模型也会自动过滤掉 password 字段
return user
这种方式确保了密码永远不会出现在 API 响应中,即使开发人员不小心返回了完整数据。
使用模型继承
更优雅的方式是使用继承:
class BaseUser(BaseModel):
"""用户基础模型"""
username: str
email: EmailStr
full_name: str | None = None
class UserIn(BaseUser):
"""用户输入模型(添加密码字段)"""
password: str
@app.post("/users/")
async def create_user(user: UserIn) -> BaseUser:
# 返回类型是 BaseUser,会自动过滤 password
# 同时编辑器也能正确理解类型关系
return user
这样做的好处:
- 编辑器和类型检查工具能正确理解类型
- FastAPI 仍然会过滤响应数据
- 代码更简洁,符合 DRY 原则
响应模型编码参数
排除未设置的字段
使用 response_model_exclude_unset=True 只返回实际设置了值的字段:
class Item(BaseModel):
name: str
description: str | None = None
price: float
tax: float = 10.5 # 有默认值
tags: list[str] = [] # 有默认值
items = {
"foo": {"name": "Foo", "price": 50.2},
"bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
}
@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)
async def read_item(item_id: str):
return items[item_id]
访问 /items/foo 返回:
{
"name": "Foo",
"price": 50.2
}
注意 description、tax 和 tags 没有出现在响应中,因为它们使用的是默认值。
访问 /items/bar 返回:
{
"name": "Bar",
"description": "The bartenders",
"price": 62,
"tax": 20.2
}
description 和 tax 被包含,因为它们被明确设置了值。
其他排除选项
# 排除默认值
@app.get("/items/{item_id}", response_model=Item, response_model_exclude_defaults=True)
# 排除 None 值
@app.get("/items/{item_id}", response_model=Item, response_model_exclude_none=True)
包含/排除特定字段
使用 response_model_include 和 response_model_exclude:
@app.get(
"/items/{item_id}/name",
response_model=Item,
response_model_include={"name", "description"} # 只包含这两个字段
)
async def read_item_name(item_id: str):
return items[item_id]
@app.get(
"/items/{item_id}/public",
response_model=Item,
response_model_exclude={"tax"} # 排除 tax 字段
)
async def read_item_public(item_id: str):
return items[item_id]
响应状态码
设置状态码
from fastapi import status
@app.post("/items/", status_code=201)
async def create_item(item: Item):
return item
@app.delete("/items/{item_id}", status_code=204)
async def delete_item(item_id: int):
return None
# 使用 status 模块(推荐,更清晰)
@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
return item
常用状态码
| 状态码 | 常量 | 含义 | 使用场景 |
|---|---|---|---|
| 200 | HTTP_200_OK | 成功 | 默认状态码 |
| 201 | HTTP_201_CREATED | 已创建 | POST 创建资源成功 |
| 204 | HTTP_204_NO_CONTENT | 无内容 | DELETE 成功,无需返回数据 |
| 400 | HTTP_400_BAD_REQUEST | 错误请求 | 客户端请求语法错误 |
| 401 | HTTP_401_UNAUTHORIZED | 未认证 | 需要登录 |
| 403 | HTTP_403_FORBIDDEN | 禁止访问 | 无权限 |
| 404 | HTTP_404_NOT_FOUND | 未找到 | 资源不存在 |
| 422 | HTTP_422_UNPROCESSABLE_ENTITY | 无法处理 | 数据验证失败 |
| 500 | HTTP_500_INTERNAL_SERVER_ERROR | 服务器错误 | 服务器内部错误 |
多种响应
多个响应模型
一个接口可能返回不同类型的响应:
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from pydantic import BaseModel
class Item(BaseModel):
name: str
price: float
class Message(BaseModel):
message: str
@app.get("/items/{item_id}", responses={
200: {"model": Item, "description": "成功返回商品"},
404: {"model": Message, "description": "商品不存在"}
})
async def read_item(item_id: int):
if item_id == 0:
return JSONResponse(
status_code=404,
content={"message": "商品不存在"}
)
return {"name": "商品名称", "price": 99.9}
直接返回 Response
在某些情况下需要完全控制响应:
from fastapi import FastAPI, Response
from fastapi.responses import JSONResponse, RedirectResponse, HTMLResponse
app = FastAPI()
@app.get("/portal")
async def get_portal(teleport: bool = False) -> Response:
if teleport:
return RedirectResponse(url="https://example.com")
return JSONResponse(content={"message": "普通响应"})
@app.get("/html/", response_class=HTMLResponse)
async def get_html():
return """
<html>
<head><title>HTML 响应</title></head>
<body><h1>Hello World</h1></body>
</html>
"""
禁用响应模型
当返回类型注解不是有效的 Pydantic 类型时,可以禁用响应模型:
@app.get("/portal", response_model=None)
async def get_portal() -> Response | dict:
# 可以返回 Response 或 dict
return {"message": "Hello"}
响应类
FastAPI 提供多种响应类:
JSONResponse
from fastapi.responses import JSONResponse
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return JSONResponse(
content={"item_id": item_id, "message": "success"},
status_code=200,
headers={"X-Custom-Header": "value"}
)
HTMLResponse
from fastapi.responses import HTMLResponse
@app.get("/html", response_class=HTMLResponse)
async def get_html():
return "<h1>Hello World</h1>"
PlainTextResponse
from fastapi.responses import PlainTextResponse
@app.get("/text", response_class=PlainTextResponse)
async def get_text():
return "Hello World"
RedirectResponse
from fastapi.responses import RedirectResponse
@app.get("/redirect")
async def redirect():
return RedirectResponse(url="/items/")
StreamingResponse
from fastapi.responses import StreamingResponse
@app.get("/stream")
async def stream():
async def generate():
for i in range(10):
yield f"data: {i}\n"
return StreamingResponse(
generate(),
media_type="text/event-stream"
)
FileResponse
from fastapi.responses import FileResponse
@app.get("/download")
async def download():
return FileResponse(
path="files/document.pdf",
filename="document.pdf",
media_type="application/pdf"
)
响应头
设置响应头
from fastapi import FastAPI, Response
app = FastAPI()
@app.get("/items/")
async def read_items(response: Response):
response.headers["X-Custom-Header"] = "Custom Value"
response.headers["Cache-Control"] = "no-cache"
return {"message": "Hello"}
在路由装饰器中设置
@app.get("/items/", headers={"X-Custom-Header": "Value"})
async def read_items():
return {"message": "Hello"}
响应模型最佳实践
1. 使用不同的输入输出模型
class UserCreate(BaseModel):
username: str
password: str
email: EmailStr
class User(BaseModel):
id: int
username: str
email: EmailStr
created_at: datetime
@app.post("/users/", response_model=User, status_code=201)
async def create_user(user: UserCreate):
# 创建用户,返回不包含密码的用户信息
pass
2. 使用模型继承减少重复
class ItemBase(BaseModel):
name: str
description: str | None = None
price: float
class ItemCreate(ItemBase):
pass
class Item(ItemBase):
id: int
created_at: datetime
3. 使用 Annotated 简化类型定义
from typing import Annotated
@app.get("/items/")
async def read_items() -> Annotated[list[Item], "商品列表"]:
return items
小结
本章我们学习了:
- 返回类型注解:声明函数返回类型
- response_model 参数:定义响应模型,过滤敏感数据
- 数据过滤:排除未设置、默认值或特定字段
- 状态码设置:自定义响应状态码
- 多种响应类型:JSON、HTML、文件、流等
- 响应头设置:自定义响应头信息
响应模型的核心价值:
- 数据验证:确保返回数据符合预期
- 安全过滤:自动过滤敏感字段
- 文档生成:自动生成 OpenAPI 文档
- 类型提示:提供更好的编辑器支持
练习
- 创建一个用户注册接口,使用不同的输入输出模型,确保密码不出现在响应中
- 实现一个接口,只返回设置了值的字段
- 创建一个返回 CSV 文件的接口
- 实现一个重定向接口,根据参数决定重定向目标