响应处理与状态码
📂 所属阶段:第一阶段 — 快速筑基(基础篇)
🔗 相关章节:路径参数与查询参数 · 请求体处理
目录
一、响应模型(Response Model)
基础实现与字段过滤
FastAPI 支持通过 response_model 声明接口的最终返回数据契约,最常用的场景是「从请求体/数据库拿到的敏感数据(如密码)自动过滤掉」。
比如下面的用户注册接口:
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
# 响应给前端的用户数据
class PublicUser(BaseModel):
id: int
name: str
email: str
# 接收注册请求的用户数据
class UserCreate(BaseModel):
name: str
email: str
password: str
@app.post("/users/", response_model=PublicUser)
async def register(user: UserCreate):
# 模拟创建用户,返回时自动抹掉 password
return {"id": 1, **user.model_dump()}
模型的核心价值
响应模型不只是「过滤字段」这么简单,它还能:
- 数据格式强制校验/转换:比如把数据库返回的
Decimal 转成 JSON 兼容的 float,或者补全缺失字段的默认值
- 自动生成规范的 Swagger/OpenAPI 文档:前端可以直接看到示例、字段说明、必填项
- 开发协作的强契约:前后端改接口必须同步模型,减少沟通成本
灵活的字段控制
除了用「请求/响应分离的模型」,还可以通过参数动态调整单个接口的字段:
from fastapi import status
# 1. 排除指定字段
@app.post("/users/", response_model=PublicUser, response_model_exclude={"email"})
async def register_hide_email(user: UserCreate):
return {"id": 1, **user.model_dump()}
# 2. 只返回已设置的字段(Pydantic 模型里有默认值但没显式赋值的会被排除)
@app.get("/users/{user_id}", response_model=PublicUser, response_model_exclude_unset=True)
async def get_user(user_id: int):
# 假设只从缓存拿到 id 和 name
return {"id": user_id, "name": "张三"}
# 3. 包含指定字段(优先级高于排除)
@app.get("/items/{item_id}", response_model=Item, response_model_include={"name", "price"})
async def get_item_summary(item_id: int):
return get_full_item_from_db(item_id)
二、状态码(Status Code)
统一风格的设置方式
FastAPI 提供了两种设置状态码的方式:
- 直接写数字(适合快速开发,但不推荐生产)
- 使用
fastapi.status 常量(有自动补全,语义清晰,生产标准)
from fastapi import status
# 直接写数字
@app.post("/items/", status_code=201)
async def create_item_raw(item: ItemCreate):
return {"id": 1, **item.model_dump()}
# 用常量(推荐)
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item_pro(item_id: int):
# 204 要求响应体为空,所以直接 return 或者 return None
return
高频状态码速查表
RESTful API 规范里常用的状态码可以记这几个:
三、直接操作 Response 对象
response_model 和 status_code 参数覆盖了 90% 的场景,但如果需要完全自定义响应细节(比如临时改状态码、加特殊响应头、设 Cookie),可以直接把 Response 或它的子类当作参数注入,或者作为返回值。
完全自定义 JSON 响应
用 fastapi.responses.JSONResponse 可以自由控制状态码、响应体和响应头:
from fastapi.responses import JSONResponse
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id > 100:
# 返回自定义的 404 响应,带额外的业务错误头
return JSONResponse(
status_code=status.HTTP_404_NOT_FOUND,
content={"code": "ITEM_NOT_FOUND", "message": "商品不存在"},
headers={"X-Error-Code": "ITEM_NOT_FOUND"}
)
# 正常情况会自动转成 JSON
return {"item_id": item_id, "name": "键盘"}
配置响应头与 Cookie
也可以注入通用的 Response 对象,这样正常返回 Pydantic 模型的同时,还能加额外信息:
from fastapi import Response
# 加响应头
@app.get("/items/")
async def list_items(response: Response):
response.headers["X-Total-Count"] = "500"
return [{"id": 1, "name": "键盘"}]
# 设 Cookie
@app.post("/login/")
async def login(response: Response):
# httponly=True 防止前端 JS 读取,减少 XSS 风险
response.set_cookie(
key="session_id",
value="secure_token_123",
httponly=True,
max_age=1800, # 30分钟过期
samesite="lax" # 防止 CSRF 攻击
)
return {"message": "登录成功"}
四、切换不同的响应类型
FastAPI 默认返回 JSON,但也支持直接通过 response_class 参数切换其他类型:
from fastapi.responses import (
HTMLResponse,
PlainTextResponse,
FileResponse,
StreamingResponse,
RedirectResponse
)
import io
# 1. 返回 HTML
@app.get("/html/", response_class=HTMLResponse)
async def get_home():
return """
<html>
<head><title>我的小站</title></head>
<body><h1>Hello FastAPI 🚀</h1></body>
</html>
"""
# 2. 返回纯文本
@app.get("/text/", response_class=PlainTextResponse)
async def get_health():
return "OK"
# 3. 返回文件(自动设置 Content-Disposition 头提示下载)
@app.get("/download/report")
async def download_report():
return FileResponse(
path="./static/report.pdf",
filename="2024年度报告.pdf", # 下载时的文件名
media_type="application/pdf"
)
# 4. 流式返回(适合大文件/实时数据)
@app.get("/stream/logs")
async def stream_logs():
async def generate_logs():
for i in range(10):
yield f"[INFO] 第 {i} 条日志\n"
return StreamingResponse(generate_logs(), media_type="text/plain")
# 5. 重定向
@app.get("/old-docs")
async def redirect_to_new_docs():
return RedirectResponse(url="/docs")
五、规范的异常处理
内置 HTTP 异常抛出
FastAPI 内置了 HTTPException,用来抛出符合规范的业务/请求异常:
from fastapi import HTTPException
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id not in items_db:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="商品不存在",
headers={"X-Error-Code": "ITEM_NOT_FOUND"} # 可选的额外响应头
)
return items_db[item_id]
业务级自定义异常
如果有特定的业务场景(比如库存不足),可以自己定义异常类,再用 @app.exception_handler() 注册处理函数:
from fastapi import Request
# 自定义异常
class OutOfStockException(Exception):
def __init__(self, item_id: int, stock: int):
self.item_id = item_id
self.stock = stock
# 注册处理函数
@app.exception_handler(OutOfStockException)
async def out_of_stock_handler(request: Request, exc: OutOfStockException):
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content={
"code": "OUT_OF_STOCK",
"message": f"商品 {exc.item_id} 库存不足,当前剩余 {exc.stock} 件"
}
)
# 抛出异常
@app.post("/items/{item_id}/buy")
async def buy_item(item_id: int):
if items_db[item_id]["stock"] < 1:
raise OutOfStockException(item_id, 0)
items_db[item_id]["stock"] -= 1
return {"message": "购买成功"}
全局兜底异常拦截
为了防止服务器内部的不可控错误(比如数据库连接失败)直接暴露给前端,可以拦截全局的 Exception:
from fastapi.exceptions import RequestValidationError
# 拦截 Pydantic 验证失败(自定义更友好的格式)
@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content={"code": "VALIDATION_ERROR", "details": exc.errors()}
)
# 拦截所有其他异常(生产环境建议不要返回 exc 的详细信息)
@app.exception_handler(Exception)
async def global_handler(request: Request, exc: Exception):
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content={"code": "INTERNAL_ERROR", "message": "服务器开小差了,请稍后重试"}
)
六、多状态码与自定义响应模型
如果一个接口可能返回多个状态码和不同的响应模型(比如正常返回数据,404 返回错误信息),可以用 responses 参数配置 Swagger 文档的显示:
from pydantic import BaseModel
# 通用成功响应
class SuccessResponse(BaseModel):
code: str = "SUCCESS"
data: dict | None = None
# 通用错误响应
class ErrorResponse(BaseModel):
code: str
message: str
@app.get(
"/items/{item_id}",
responses={
200: {"model": SuccessResponse, "description": "查询成功"},
404: {"model": ErrorResponse, "description": "商品不存在"}
}
)
async def read_item(item_id: int):
if item_id not in items_db:
raise HTTPException(
status_code=404,
detail={"code": "ITEM_NOT_FOUND", "message": "商品不存在"}
)
return {"code": "SUCCESS", "data": items_db[item_id]}
七、通用分页响应方案
几乎所有的列表接口都需要分页,FastAPI 结合 Pydantic 的泛型可以写一个通用的分页模型:
from typing import Generic, List, TypeVar
# 定义泛型 T,代表任意数据类型
T = TypeVar("T")
# 通用分页响应模型
class PaginatedResponse(BaseModel, Generic[T]):
items: List[T] # 当前页的数据列表
total: int # 总条数
page: int # 当前页码
page_size: int # 每页条数
total_pages: int # 总页数
# 使用示例:查询商品列表
@app.get("/items/", response_model=PaginatedResponse[Item])
async def list_items(page: int = 1, page_size: int = 10):
# 限制每页最大条数防止性能问题
page_size = min(page_size, 100)
# 模拟数据库查询
all_items = list(items_db.values())
start = (page - 1) * page_size
end = start + page_size
# 计算总页数
total_pages = (len(all_items) + page_size - 1) // page_size
return {
"items": all_items[start:end],
"total": len(all_items),
"page": page,
"page_size": page_size,
"total_pages": total_pages
}
八、小结
这篇文章覆盖了 FastAPI 响应处理的核心功能,整理成速查表方便记忆: