响应处理与状态码

📂 所属阶段:第一阶段 — 快速筑基(基础篇)
🔗 相关章节:路径参数与查询参数 · 请求体处理

目录


一、响应模型(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()}

模型的核心价值

响应模型不只是「过滤字段」这么简单,它还能:

  1. 数据格式强制校验/转换:比如把数据库返回的 Decimal 转成 JSON 兼容的 float,或者补全缺失字段的默认值
  2. 自动生成规范的 Swagger/OpenAPI 文档:前端可以直接看到示例、字段说明、必填项
  3. 开发协作的强契约:前后端改接口必须同步模型,减少沟通成本

灵活的字段控制

除了用「请求/响应分离的模型」,还可以通过参数动态调整单个接口的字段

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 提供了两种设置状态码的方式:

  1. 直接写数字(适合快速开发,但不推荐生产)
  2. 使用 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 规范里常用的状态码可以记这几个:

状态码对应常量说明(RESTful 语义)
200HTTP_200_OK查询/更新/部分更新成功
201HTTP_201_CREATED创建资源成功(比如 POST 注册)
204HTTP_204_NO_CONTENT删除成功/空查询响应成功
400HTTP_400_BAD_REQUEST请求参数格式/逻辑错误
401HTTP_401_UNAUTHORIZED未登录/Token 无效
403HTTP_403_FORBIDDEN已登录但无权限操作该资源
404HTTP_404_NOT_FOUND请求的资源不存在
422HTTP_422_UNPROCESSABLE_ENTITYFastAPI/Pydantic 自动校验失败
500HTTP_500_INTERNAL_SERVER_ERROR服务器内部不可控错误

三、直接操作 Response 对象

response_modelstatus_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": "键盘"}

也可以注入通用的 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 响应处理的核心功能,整理成速查表方便记忆:

功能场景核心用法/类/参数
声明响应契约+过滤字段response_model=PublicUser
灵活调整单个接口字段response_model_exclude/include/unset
统一语义的状态码fastapi.status.HTTP_201_CREATED
完全自定义 JSON 响应fastapi.responses.JSONResponse
加响应头/设 Cookie注入 Response 参数
切换响应类型response_class=HTMLResponse
抛出规范业务异常raise HTTPException
自定义业务异常定义类 + @app.exception_handler()
全局兜底拦截拦截 Exception
多状态码+多模型文档responses={200: {"model": Success}}
通用分页Pydantic 泛型 PaginatedResponse[T]