FastAPI 学习之响应模型

在 FastAPI 中，响应模型（Response Model） 用于精确控制 API 返回的数据结构和文档生成，通过 Pydantic 模型实现数据过滤、格式转换和安全防护。

响应模型的作用

数据过滤：仅返回模型中定义的字段，隐藏敏感或不必要的数据。
数据验证：确保响应数据符合模型定义的类型和约束。
文档生成：自动在 Swagger UI 中展示响应结构。
序列化：将复杂数据类型（如 ORM 对象）转换为 JSON 兼容格式。

基础响应控制

基本用法

通过 response_model 参数指定响应模型。

示例：基本响应模型

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# 请求体模型（输入）
class ItemInput(BaseModel):
    name: str
    price: float
    tax: float = 10.0

# 响应模型（输出）
class ItemOutput(BaseModel):
    name: str
    price: float

@app.post("/items/", response_model=ItemOutput)
def create_item(item: ItemInput):
    # 返回的数据会自动过滤掉未在 ItemOutput 中定义的字段（如 tax）
    return item

效果：

输入数据包含name、price、tax，但响应中仅返回 name 和 price。
Swagger UI 中会显示响应结构为ItemOutput。

自动文档生成

class ItemResponse(BaseModel):
    id: int
    name: str = Field(example="MacBook Pro")
    price: float = Field(description="人民币价格", gt=0)

@app.get("/items/{id}", response_model=ItemResponse)
async def get_item(id: int):
    ...

Swagger 效果：自动显示字段示例和描述

响应状态码

FastAPI 允许你灵活地控制 HTTP 状态码，用于明确表示 API 请求的结果（如成功、错误、重定向等）。以下是状态码的详细说明和用法：

默认状态码

FastAPI 根据 HTTP 方法自动设置默认状态码：

GET:200 OK
POST:201 Created
PUT:200 OK
DELETE:204 No Content

示例：显式指定默认状态码

from fastapi import FastAPI

app = FastAPI()

@app.post("/items/", status_code=201)  # 显式设置 201
def create_item():
    return {"message": "Item created"}

常用 HTTP 状态码

状态码	名称	适用场景
200	OK	通用成功响应（如 GET 或 PUT 成功）。
201	Created	资源创建成功（如 POST 创建新数据）。
204	No Content	成功但无返回内容（如 DELETE 成功）。
400	Bad Request	客户端请求错误（如参数校验失败）。
401	Unauthorized	未认证（如未提供 Token）。
403	Forbidden	无权限访问资源（如用户角色不符）。
404	Not Found	资源不存在（如请求的 ID 在数据库中未找到）。
422	Unprocessable Entity	请求体或参数格式正确但语义错误（由 Pydantic 自动触发）。
500	Internal Server Error	服务器内部错误（如未捕获的异常）。

动态设置状态码

在路由处理函数中，可以通过返回 Response 子类（如 JSONResponse）动态设置状态码。

示例：根据条件返回不同状态码

from fastapi import FastAPI, status
from fastapi.responses import JSONResponse

app = FastAPI()

@app.post("/items/")
def create_item(item_id: int):
    if item_id < 1:
        return JSONResponse(
            status_code=status.HTTP_400_BAD_REQUEST,
            content={"message": "Item ID 必须大于 0"}
        )
    return {"item_id": item_id}

使用 HTTPException 抛出错误状态码

通过抛出 HTTPException 快速返回错误状态码和详细信息。

示例：资源未找到时返回 404

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"1": "Apple", "2": "Banana"}

@app.get("/items/{item_id}")
def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "Invalid ID"}  # 可选自定义响应头
        )
    return {"item": items[item_id]}

自定义状态码与响应模型

结合 responses 参数，为不同状态码指定响应模型（在 Swagger UI 中展示）。

示例：定义成功和错误响应

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str

class ErrorMessage(BaseModel):
    code: int
    message: str

@app.post("/items/", responses={
    201: {"model": Item, "description": "成功创建"},
    400: {"model": ErrorMessage, "description": "无效输入"}
})
def create_item(item: Item):
    if item.name == "invalid":
        raise HTTPException(status_code=400, detail={"code": 400, "message": "名称无效"})
    return item

状态码工具类

FastAPI 提供 status 模块，包含预定义的状态码常量，避免手动记忆数字。

示例：使用 status 模块

from fastapi import FastAPI, status

app = FastAPI()

@app.post("/items/", status_code=status.HTTP_201_CREATED)
def create_item():
    return {"message": "Item created"}

重定向状态码

使用 RedirectResponse 实现 3xx 重定向。

示例：永久重定向（301）

from fastapi import FastAPI
from fastapi.responses import RedirectResponse

app = FastAPI()

@app.get("/old-url")
def redirect():
    return RedirectResponse(url="/new-url", status_code=301)

状态码最佳实践

语义化：选择符合操作结果的状态码（如201 用于资源创建）。
一致性：相同类型的操作使用相同状态码（如所有 POST 创建返回201）。
错误处理：
- 客户端错误使用4xx（如 400 参数错误，404 资源不存在）。
- 服务端错误使用5xx（如 500 未捕获异常）。
文档化：通过responses 参数在 Swagger UI 中明确展示可能的状态码。

高级响应策略

多状态码响应模型

结合 status_code 参数，定义不同状态码对应的响应模型。

from fastapi.responses import JSONResponse

@app.get("/secret-data/",
    response_model=SecretData,
    responses={
        200: {"model": SecretData, "description": "成功获取"},
        403: {"model": ErrorMsg, "content": {"text/plain": {"example": "无权限访问"}}}
    }
)
async def get_secret():
    if not has_permission():
        return JSONResponse(
            status_code=403,
            content={"error": "无权限访问"}
        )
    return SecretData(...)

动态字段过滤

@app.get("/users/me", 
    response_model=UserResponse,
    response_model_include={"username", "email"},
    response_model_exclude_unset=True
)
async def get_current_user():
    ...

请求结果：仅返回指定的username 和 email 字段

自定义响应模型配置

通过 Pydantic 的 Config 类自定义模型行为。

示例：自动转换字段名

class User(BaseModel):
    name: str
    created_at: datetime

    class Config:
        json_encoders = {
            datetime: lambda dt: dt.isoformat()  # 自定义 datetime 序列化
        }
        allow_population_by_field_name = True  # 允许通过别名访问字段

响应序列化

FastAPI 的响应序列化（Response Serialization）是将 Python 对象转换为客户端可接收的格式（如 JSON）的过程。通过 Pydantic 模型和 FastAPI 的内置机制，可以灵活控制响应数据的结构和格式。

FastAPI 的序列化流程：

模型转换：将 Python 对象（如 Pydantic 模型、ORM 对象）转换为字典。
字段过滤：根据响应模型（response_model）排除未定义的字段。
JSON 序列化：将字典转换为 JSON 字符串（使用dumps）。
编码处理：处理特殊类型（如datetime、UUID）。

处理复杂数据类型

日期时间序列化

Pydantic 自动将 datetime 转换为 ISO 格式字符串：

from datetime import datetime

class Post(BaseModel):
    title: str
    created_at: datetime

@app.get("/posts/", response_model=Post)
def get_post():
    return {"title": "Hello", "created_at": datetime.now()}

响应：{“title”: “Hello”, “created_at”: “2023-09-01T12:34:56.789Z”}

自定义 JSON 编码器

通过 json_encoders 处理特殊类型（如自定义类）：

from pydantic import BaseModel

class CustomObject:
    def __init__(self, value: str):
        self.value = value

class Item(BaseModel):
    obj: CustomObject

    class Config:
        json_encoders = {
            CustomObject: lambda v: {"value": v.value}  # 自定义序列化逻辑
        }

@app.get("/items/", response_model=Item)
def get_item():
    return Item(obj=CustomObject("test"))

响应：{“obj”: {“value”: “test”}}

分页响应模型

class PaginatedResponse(BaseModel):
    data: list[User]
    total: int
    page: int

@app.get("/users/list", response_model=PaginatedResponse)
def list_users(page: int = 1):
    users = [{"name": "Alice"}, {"name": "Bob"}]
    return {"data": users, "total": 2, "page": page}

响应序列化配置

通过 Pydantic 的 Config 类自定义模型行为：

class User(BaseModel):
    name: str
    age: int

    class Config:
        json_schema_extra = {
            "example": {
                "name": "Alice",
                "age": 30
            }
        }
        allow_population_by_field_name = True  # 允许通过别名填充字段

常见问题与解决方案

问题	解决方案
敏感字段泄露	使用 response_model 排除字段
日期时间格式不一致	自定义 json_encoders 或使用 datetime.isoformat()
ORM 对象无法序列化	启用 orm_mode = True
循环引用导致序列化失败	使用 jsonable_encoder 或重构模型
性能瓶颈	避免深度嵌套模型，使用 response_model_by_alias=False

安全防护机制

敏感数据过滤

class SafeUserResponse(BaseModel):
    id: int
    username: str
    # 不包含 password 字段

class UserDB(BaseModel):
    id: int
    username: str
    password_hash: str

@app.post("/login/", response_model=SafeUserResponse)
async def login():
    user = UserDB(...)  # 包含敏感字段的数据库对象
    return user  # 自动过滤 password_hash

深度嵌套过滤

响应模型可以嵌套其他模型，处理复杂数据结构。

示例：嵌套响应模型

class Address(BaseModel):
    city: str
    street: str

class UserWithAddress(BaseModel):
    name: str
    address: Address

@app.get("/users/{user_id}", response_model=UserWithAddress)
def get_user(user_id: int):
    return {
        "name": "Alice",
        "address": {
            "city": "Beijing",
            "street": "长安街"
        }
    }

性能优化技巧

使用 response_model_by_alias

禁用别名处理以加速序列化：

@app.get("/items/", response_model=Item, response_model_by_alias=False)
def get_item():
    return Item(name="Apple")

预序列化数据

直接返回字典或 Pydantic 模型实例，避免多次转换：

@app.get("/items/")
def get_item() -> dict:  # 直接返回字典
    return {"name": "Apple", "price": 10.0}

ORM 模式加速

当从数据库（如 SQLAlchemy）返回 ORM 对象时，需通过响应模型将其转换为 Pydantic 模型。

示例：ORM 到 Pydantic 的转换

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
from pydantic import BaseModel

Base = declarative_base()

# SQLAlchemy 模型
class UserDB(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    hashed_password = Column(String)  # 敏感字段，不应返回

# Pydantic 响应模型
class UserOut(BaseModel):
    id: int
    name: str

    class Config:
        orm_mode = True  # 允许从 ORM 对象转换

@app.get("/users/{user_id}", response_model=UserOut)
def get_user(user_id: int):
    user = session.query(UserDB).filter(UserDB.id == user_id).first()
    return user  # 自动过滤 hashed_password 字段

关键点：

orm_mode = True允许 Pydantic 模型从 ORM 对象读取数据。
响应模型仅包含id 和 name，隐藏了 hashed_password。

响应缓存

from fastapi_cache.decorator import cache

@app.get("/products/", response_model=list[ProductResponse])
@cache(expire=60)  # 缓存60秒
async def get_products():
...

处理循环引用

使用 jsonable_encoder 手动处理复杂对象：

from fastapi.encoders import jsonable_encoder

class User(BaseModel):
    name: str
    friends: list["User"] = []  # 循环引用

User.update_forward_refs()  # 解决前向引用

@app.get("/users/")
def get_user():
    user = User(name="Alice")
    user.friends.append(User(name="Bob"))
    return jsonable_encoder(user)  # 手动序列化

复杂场景处理

条件响应模型

一个接口可以根据条件返回不同的模型（如不同用户角色返回不同字段）。

示例：联合类型（Union）

from typing import Union
from pydantic import BaseModel

class AdminUser(BaseModel):
    name: str
    role: str = "admin"
    permissions: list[str]

class NormalUser(BaseModel):
    name: str
    role: str = "user"

@app.get("/users/{user_id}", response_model=Union[AdminUser, NormalUser])
def get_user(user_id: int):
    if user_id == 1:
        return AdminUser(name="Alice", permissions=["read", "write"])
    else:
        return NormalUser(name="Bob")

效果：

用户 1 返回AdminUser 结构，其他用户返回 NormalUser。
Swagger UI 会展示两种可能的响应模型。

文件流响应

from fastapi.responses import StreamingResponse

@app.get("/download/{filename}", 
    responses={
        200: {
            "content": {"application/octet-stream": {}},
            "description": "文件下载"
        }
    }
)
async def download_file(filename: str):
    def iter_file():
        with open(f"storage/{filename}", "rb") as f:
            yield from f
            
    return StreamingResponse(iter_file(), media_type="application/octet-stream")

最佳实践总结

分层设计模型

模型类型	用途	示例
Input Model	接收请求数据	UserCreate
Database Model	数据库交互模型	UserDB
Response Model	控制输出数据	UserResponse
Update Model	部分更新专用（Optional 字段）	UserUpdate

安全规范

class StrictResponse(BaseModel):
    class Config:
        extra = "forbid"  # 禁止额外字段
        anystr_strip_whitespace = True  # 自动去除空格
        json_encoders = {
            datetime: lambda v: v.isoformat()  # 统一时间格式
        }

文档增强

class OpenAPIExample:
    """集中管理响应示例"""
    USER_EXAMPLE = {
        "application/json": {
            "example": {
                "id": 1,
                "username": "fastapi-user",
                "email": "user@example.com"
            }
        }
    }

@app.get("/users/{id}", 
    response_model=UserResponse,
    responses={200: OpenAPIExample.USER_EXAMPLE}
)
async def get_user(id: int):
    ...

通过合理使用响应模型，开发者可以实现：

数据脱敏：自动过滤数据库中的敏感字段
文档一致性：保证 Swagger 文档与实际返回数据结构一致
性能优化：通过 ORM 模式减少数据转换开销
灵活控制：动态决定返回字段和数据结构

结合请求验证和路由依赖注入，可构建出安全、高效且易维护的 API 服务。