跳到主要内容

响应模型

响应模型定义了 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
}

注意 descriptiontaxtags 没有出现在响应中,因为它们使用的是默认值。

访问 /items/bar 返回:

{
"name": "Bar",
"description": "The bartenders",
"price": 62,
"tax": 20.2
}

descriptiontax 被包含,因为它们被明确设置了值。

其他排除选项

# 排除默认值
@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_includeresponse_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

常用状态码

状态码常量含义使用场景
200HTTP_200_OK成功默认状态码
201HTTP_201_CREATED已创建POST 创建资源成功
204HTTP_204_NO_CONTENT无内容DELETE 成功,无需返回数据
400HTTP_400_BAD_REQUEST错误请求客户端请求语法错误
401HTTP_401_UNAUTHORIZED未认证需要登录
403HTTP_403_FORBIDDEN禁止访问无权限
404HTTP_404_NOT_FOUND未找到资源不存在
422HTTP_422_UNPROCESSABLE_ENTITY无法处理数据验证失败
500HTTP_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

小结

本章我们学习了:

  1. 返回类型注解:声明函数返回类型
  2. response_model 参数:定义响应模型,过滤敏感数据
  3. 数据过滤:排除未设置、默认值或特定字段
  4. 状态码设置:自定义响应状态码
  5. 多种响应类型:JSON、HTML、文件、流等
  6. 响应头设置:自定义响应头信息

响应模型的核心价值:

  • 数据验证:确保返回数据符合预期
  • 安全过滤:自动过滤敏感字段
  • 文档生成:自动生成 OpenAPI 文档
  • 类型提示:提供更好的编辑器支持

练习

  1. 创建一个用户注册接口,使用不同的输入输出模型,确保密码不出现在响应中
  2. 实现一个接口,只返回设置了值的字段
  3. 创建一个返回 CSV 文件的接口
  4. 实现一个重定向接口,根据参数决定重定向目标