第7章 - 响应模型

嗨，朋友！我是长安。

这一章我们来讲响应模型。说实话，这是我工作中用得非常多的一个功能，它能帮我们保护敏感数据，比如密码。当年因为不知道这个，我差点把用户密码都返回给前端了！

🎯 本章目标

理解响应模型的作用
学会定义和使用响应模型
掌握响应数据的过滤
学会设置响应状态码

1️⃣ 为什么需要响应模型？

假设我们有一个用户模型：

class User(BaseModel):
    id: int
    username: str
    password: str  # 敏感信息！
    email: str

如果直接返回，密码也会被返回，这是不安全的！

响应模型可以帮我们：

过滤敏感数据 - 不返回密码等敏感信息
规范返回格式 - 确保返回数据格式一致
自动文档 - 在 API 文档中显示返回数据结构

2️⃣ 定义响应模型

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# 请求模型（包含密码）
class UserCreate(BaseModel):
    username: str
    password: str
    email: str

# 响应模型（不包含密码）
class UserResponse(BaseModel):
    id: int
    username: str
    email: str

# 模拟数据库
fake_db = {}
user_id_counter = 0

@app.post("/users", response_model=UserResponse)
def create_user(user: UserCreate):
    global user_id_counter
    user_id_counter += 1
    
    # 存储到"数据库"（包含密码）
    new_user = {
        "id": user_id_counter,
        "username": user.username,
        "password": user.password,  # 存储密码
        "email": user.email
    }
    fake_db[user_id_counter] = new_user
    
    # 返回时，FastAPI 会自动过滤掉 password
    return new_user

即使 return new_user 包含 password，由于设置了 response_model=UserResponse，返回给客户端的数据不会包含 password！

3️⃣ response_model 的作用

数据过滤 - 只返回模型中定义的字段
数据验证 - 验证返回数据是否符合模型
数据转换 - 自动进行类型转换
文档生成 - 在 Swagger UI 中显示响应结构

4️⃣ 排除和包含字段

response_model_exclude

排除某些字段：

class User(BaseModel):
    id: int
    username: str
    password: str
    email: str
    created_at: str

@app.get("/users/{user_id}", response_model=User, response_model_exclude={"password"})
def get_user(user_id: int):
    return fake_db.get(user_id)

response_model_include

只包含某些字段：

@app.get("/users/{user_id}/brief", response_model=User, response_model_include={"id", "username"})
def get_user_brief(user_id: int):
    return fake_db.get(user_id)

5️⃣ 排除默认值和 None

response_model_exclude_unset

排除未设置的字段（使用默认值的字段）：

class Item(BaseModel):
    name: str
    price: float
    description: str = "暂无描述"
    tax: float = 0.0

@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)
def get_item(item_id: int):
    return {"name": "苹果", "price": 5.0}
    # 只返回 {"name": "苹果", "price": 5.0}
    # 不会返回 description 和 tax

response_model_exclude_none

排除值为 None 的字段：

@app.get("/items/{item_id}", response_model=Item, response_model_exclude_none=True)
def get_item(item_id: int):
    return {"name": "苹果", "price": 5.0, "description": None}
    # description 为 None，不会返回

6️⃣ 返回列表

from typing import List

class User(BaseModel):
    id: int
    username: str
    email: str

@app.get("/users", response_model=List[User])
def get_users():
    return [
        {"id": 1, "username": "张三", "email": "zhangsan@example.com"},
        {"id": 2, "username": "李四", "email": "lisi@example.com"},
    ]

7️⃣ 通用响应模型

实际项目中，通常会定义统一的响应格式：

from typing import Generic, TypeVar, Optional, List
from pydantic import BaseModel

T = TypeVar('T')

class Response(BaseModel, Generic[T]):
    code: int = 200
    message: str = "success"
    data: Optional[T] = None

class UserOut(BaseModel):
    id: int
    username: str
    email: str

# 单个用户响应
@app.get("/users/{user_id}", response_model=Response[UserOut])
def get_user(user_id: int):
    user = {"id": user_id, "username": "张三", "email": "test@example.com"}
    return Response(data=user)

# 用户列表响应
@app.get("/users", response_model=Response[List[UserOut]])
def get_users():
    users = [
        {"id": 1, "username": "张三", "email": "zhangsan@example.com"},
        {"id": 2, "username": "李四", "email": "lisi@example.com"},
    ]
    return Response(data=users)

返回格式：

{
    "code": 200,
    "message": "success",
    "data": {
        "id": 1,
        "username": "张三",
        "email": "zhangsan@example.com"
    }
}

8️⃣ 设置响应状态码

默认状态码

from fastapi import status

@app.post("/users", status_code=status.HTTP_201_CREATED)
def create_user(user: UserCreate):
    return {"message": "创建成功"}

常用状态码

状态码	常量	含义
200	HTTP_200_OK	成功
201	HTTP_201_CREATED	创建成功
204	HTTP_204_NO_CONTENT	成功但无内容
400	HTTP_400_BAD_REQUEST	请求错误
401	HTTP_401_UNAUTHORIZED	未授权
403	HTTP_403_FORBIDDEN	禁止访问
404	HTTP_404_NOT_FOUND	未找到
500	HTTP_500_INTERNAL_SERVER_ERROR	服务器错误

动态状态码

from fastapi import Response

@app.get("/items/{item_id}")
def get_item(item_id: int, response: Response):
    if item_id not in fake_db:
        response.status_code = status.HTTP_404_NOT_FOUND
        return {"error": "商品不存在"}
    return fake_db[item_id]

9️⃣ 多种响应模型

一个接口可能返回不同的响应：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    id: int
    username: str

class ErrorResponse(BaseModel):
    error: str
    detail: str

@app.get(
    "/users/{user_id}",
    response_model=User,
    responses={
        404: {"model": ErrorResponse, "description": "用户不存在"},
        500: {"model": ErrorResponse, "description": "服务器错误"}
    }
)
def get_user(user_id: int):
    if user_id not in fake_db:
        raise HTTPException(status_code=404, detail="用户不存在")
    return fake_db[user_id]

🔟 完整示例

from typing import List, Optional, Generic, TypeVar
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field

app = FastAPI(title="响应模型示例")

T = TypeVar('T')

# ========== 通用响应模型 ==========

class BaseResponse(BaseModel, Generic[T]):
    """通用响应模型"""
    code: int = Field(default=200, description="状态码")
    message: str = Field(default="success", description="提示信息")
    data: Optional[T] = Field(default=None, description="响应数据")

class PageData(BaseModel, Generic[T]):
    """分页数据模型"""
    items: List[T]
    total: int
    page: int
    size: int

# ========== 用户模型 ==========

class UserCreate(BaseModel):
    """创���用户请求"""
    username: str = Field(min_length=3, max_length=20)
    password: str = Field(min_length=6)
    email: str
    phone: Optional[str] = None

class UserOut(BaseModel):
    """用户响应（不含密码）"""
    id: int
    username: str
    email: str
    phone: Optional[str] = None
    is_active: bool = True

class UserDetail(UserOut):
    """用户详情（包含更多信息）"""
    created_at: str
    updated_at: str
    login_count: int

# ========== 模拟数据 ==========

fake_users_db = {
    1: {
        "id": 1,
        "username": "zhangsan",
        "password": "hashed_password",
        "email": "zhangsan@example.com",
        "phone": "13800138001",
        "is_active": True,
        "created_at": "2023-01-01 10:00:00",
        "updated_at": "2023-12-01 15:30:00",
        "login_count": 100
    },
    2: {
        "id": 2,
        "username": "lisi",
        "password": "hashed_password",
        "email": "lisi@example.com",
        "phone": None,
        "is_active": True,
        "created_at": "2023-06-15 08:00:00",
        "updated_at": "2023-11-20 12:00:00",
        "login_count": 50
    }
}

# ========== 接口定义 ==========

@app.post(
    "/users",
    response_model=BaseResponse[UserOut],
    status_code=status.HTTP_201_CREATED,
    tags=["用户管理"],
    summary="创建用户"
)
def create_user(user: UserCreate):
    """创建新用户，返回用户信息（不含密码）"""
    new_id = max(fake_users_db.keys()) + 1
    new_user = {
        "id": new_id,
        "username": user.username,
        "password": user.password,  # 实际应该加密
        "email": user.email,
        "phone": user.phone,
        "is_active": True,
        "created_at": "2023-12-01 10:00:00",
        "updated_at": "2023-12-01 10:00:00",
        "login_count": 0
    }
    fake_users_db[new_id] = new_user
    
    return BaseResponse(
        code=201,
        message="用户创建成功",
        data=new_user  # password 会被自动过滤
    )

@app.get(
    "/users",
    response_model=BaseResponse[PageData[UserOut]],
    tags=["用户管理"],
    summary="获取用户列表"
)
def get_users(page: int = 1, size: int = 10):
    """获取用户列表，支持分页"""
    users = list(fake_users_db.values())
    total = len(users)
    start = (page - 1) * size
    end = start + size
    items = users[start:end]
    
    return BaseResponse(
        data=PageData(
            items=items,
            total=total,
            page=page,
            size=size
        )
    )

@app.get(
    "/users/{user_id}",
    response_model=BaseResponse[UserDetail],
    tags=["用户管理"],
    summary="获取用户详情"
)
def get_user(user_id: int):
    """获取单个用户的详细信息"""
    if user_id not in fake_users_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="用户不存在"
        )
    return BaseResponse(data=fake_users_db[user_id])

@app.get(
    "/users/{user_id}/brief",
    response_model=UserOut,
    response_model_include={"id", "username"},
    tags=["用户管理"],
    summary="获取用户简要信息"
)
def get_user_brief(user_id: int):
    """只返回用户ID和用户名"""
    if user_id not in fake_users_db:
        raise HTTPException(status_code=404, detail="用户不存在")
    return fake_users_db[user_id]

@app.delete(
    "/users/{user_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    tags=["用户管理"],
    summary="删除用户"
)
def delete_user(user_id: int):
    """删除用户，成功返回204无内容"""
    if user_id not in fake_users_db:
        raise HTTPException(status_code=404, detail="用户不存在")
    del fake_users_db[user_id]
    return None

📝 小结

本章我们学习了：

✅ 响应模型的作用��定义
✅ 使用 response_model 过滤数据
✅ response_model_exclude 和 response_model_include
✅ 排除默认值和 None
✅ 通用响应模型的设计
✅ 设置响应状态码

🏃 下一步

现在你已经掌握了请求和响应的处理，接下来学习如何实现完整的 CRUD 操作！

👉 第8章 - CRUD 操作

💪 练习题

创建一个商品响应模型，不包含成本价（cost）字段
设计一个统一的 API 响应格式，包含 code、message、data
创建一个接口，成功返回 201，失败返回 400

参考答案

from typing import Optional, Generic, TypeVar
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel

app = FastAPI()
T = TypeVar('T')

# 练习1
class ProductInternal(BaseModel):
    id: int
    name: str
    price: float
    cost: float  # 成本价

class ProductOut(BaseModel):
    id: int
    name: str
    price: float
    # 不包含 cost

@app.get("/products/{product_id}", response_model=ProductOut)
def get_product(product_id: int):
    return {"id": 1, "name": "苹果", "price": 5.0, "cost": 2.0}

# 练习2
class ApiResponse(BaseModel, Generic[T]):
    code: int = 200
    message: str = "success"
    data: Optional[T] = None

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

@app.get("/users/{user_id}", response_model=ApiResponse[User])
def get_user(user_id: int):
    return ApiResponse(data={"id": user_id, "name": "张三"})

# 练习3
class ItemCreate(BaseModel):
    name: str
    price: float

@app.post("/items", status_code=status.HTTP_201_CREATED)
def create_item(item: ItemCreate):
    if item.price < 0:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="价格不能为负数"
        )
    return {"message": "创建成功", "item": item}