第6章 - 请求体

嗨，朋友！我是长安。

这一章我们来讲请求体，也就是如何接收客户端发来的 JSON 数据。这是 FastAPI 最强大的功能之一，当年第一次使用时真的被惊艳到了！

🎯 本章目标

  • 理解什么是请求体
  • 学会使用 Pydantic 定义数据模型
  • 掌握请求体的数据验证
  • 学会处理嵌套模型

1️⃣ 什么是请求体？

请求体（Request Body）是客户端发送给服务器的数据，通常用于 POST、PUT、PATCH 请求。

比如创建用户时，需要发送用户信息：

{
    "name": "张三",
    "age": 20,
    "email": "zhangsan@example.com"
}

这个 JSON 数据就是请求体。

2���⃣ 使用 Pydantic 定义模型

FastAPI 使用 Pydantic 来定义和验证数据模型：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# 定义数据模型
class User(BaseModel):
    name: str
    age: int
    email: str

@app.post("/users")
def create_user(user: User):
    return {
        "message": "用户创建成功",
        "user": user
    }

测试接口

在 Swagger UI（/docs）中测试，或使用 curl：

curl -X POST "http://127.0.0.1:8000/users" \
     -H "Content-Type: application/json" \
     -d '{"name": "张三", "age": 20, "email": "zhangsan@example.com"}'

返回：

{
    "message": "用户创建成功",
    "user": {
        "name": "张三",
        "age": 20,
        "email": "zhangsan@example.com"
    }
}

3️⃣ 自动数据验证

Pydantic 会自动验证数据类型：

class User(BaseModel):
    name: str    # 必须是字符串
    age: int     # 必须是整数
    email: str   # 必须是字符串

如果传入错误的数据：

{
    "name": "张三",
    "age": "二十",  // 应该是数字
    "email": "test@example.com"
}

FastAPI 会返回清晰的错误信息：

{
    "detail": [
        {
            "type": "int_parsing",
            "loc": ["body", "age"],
            "msg": "Input should be a valid integer, unable to parse string as an integer",
            "input": "二十"
        }
    ]
}

4️⃣ 可选字段和默认值

from typing import Optional
from pydantic import BaseModel

class User(BaseModel):
    name: str                           # 必填
    age: int                            # 必填
    email: Optional[str] = None         # 可选，默认 None
    is_active: bool = True              # 可选，默认 True
    role: str = "user"                  # 可选，默认 "user"

现在只需要传 name 和 age：

{
    "name": "张三",
    "age": 20
}

返回：

{
    "name": "张三",
    "age": 20,
    "email": null,
    "is_active": true,
    "role": "user"
}

5️⃣ 字段验证

使用 Field 进行更详细的验证：

from pydantic import BaseModel, Field

class User(BaseModel):
    name: str = Field(
        min_length=2,
        max_length=20,
        description="用户名"
    )
    age: int = Field(
        ge=0,
        le=150,
        description="年龄"
    )
    email: str = Field(
        pattern=r"^[\w\.-]+@[\w\.-]+\.\w+$",
        description="邮箱地址"
    )
    score: float = Field(
        default=0,
        ge=0,
        le=100,
        description="分数"
    )

Field 常用参数

参数说明示例
default默认值default=0
min_length最小长度min_length=2
max_length最大长度max_length=50
ge大于等于ge=0
gt大于gt=0
le小于等于le=100
lt小于lt=100
pattern正则表达式pattern=r"^\d+$"
description描述description="用户名"
example示例值example="张三"

6️⃣ 嵌套模型

模型可以嵌套使用：

from typing import List, Optional
from pydantic import BaseModel

# 地址模型
class Address(BaseModel):
    province: str
    city: str
    street: str

# 用户模型（包含地址）
class User(BaseModel):
    name: str
    age: int
    address: Address  # 嵌套模型

@app.post("/users")
def create_user(user: User):
    return user

请求数据：

{
    "name": "张三",
    "age": 20,
    "address": {
        "province": "广东省",
        "city": "深圳市",
        "street": "科技园路100号"
    }
}

列表嵌套

class OrderItem(BaseModel):
    product_id: int
    quantity: int
    price: float

class Order(BaseModel):
    order_no: str
    items: List[OrderItem]  # 订单项列表
    total: float

@app.post("/orders")
def create_order(order: Order):
    return order

请求数据：

{
    "order_no": "ORD001",
    "items": [
        {"product_id": 1, "quantity": 2, "price": 10.0},
        {"product_id": 2, "quantity": 1, "price": 20.0}
    ],
    "total": 40.0
}

7️⃣ 请求体 + 路径参数 + 查询参数

可以同时使用三种参数：

from typing import Optional
from fastapi import FastAPI, Query
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    description: Optional[str] = None

@app.put("/items/{item_id}")
def update_item(
    item_id: int,                              # 路径参数
    item: Item,                                # 请求体
    notify: bool = Query(default=False)        # 查询参数
):
    return {
        "item_id": item_id,
        "item": item,
        "notify": notify
    }

FastAPI 会自动识别：

  • item_id 在路径中定义 → 路径参数
  • item 是 Pydantic 模型 → 请求体
  • notify 是简单类型且不在路径中 → 查询参数

8️⃣ 多个请求体

可以接收多个请求体：

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

class Item(BaseModel):
    name: str
    price: float

@app.post("/create")
def create(user: User, item: Item):
    return {"user": user, "item": item}

请求数据需要这样组织：

{
    "user": {
        "name": "张三",
        "age": 20
    },
    "item": {
        "name": "苹果",
        "price": 5.0
    }
}

9️⃣ 使用 Body 嵌入单个值

有时候需要在请求体中嵌入单个值：

from fastapi import Body

@app.post("/items")
def create_item(
    name: str = Body(...),
    price: float = Body(...),
    description: str = Body(default=None)
):
    return {
        "name": name,
        "price": price,
        "description": description
    }

请求数据：

{
    "name": "苹果",
    "price": 5.0,
    "description": "新鲜水果"
}

🔟 完整示例

from typing import List, Optional
from fastapi import FastAPI, Body, Query
from pydantic import BaseModel, Field

app = FastAPI(title="请求体示例")

# ========== 模型定义 ==========

class Address(BaseModel):
    """地址模型"""
    province: str = Field(description="省份")
    city: str = Field(description="城市")
    detail: str = Field(description="详细地址")

class UserCreate(BaseModel):
    """创建用户请求模型"""
    username: str = Field(
        min_length=3,
        max_length=20,
        description="用户名",
        example="zhangsan"
    )
    password: str = Field(
        min_length=6,
        max_length=20,
        description="密码"
    )
    email: str = Field(
        pattern=r"^[\w\.-]+@[\w\.-]+\.\w+$",
        description="邮箱",
        example="zhangsan@example.com"
    )
    age: Optional[int] = Field(
        default=None,
        ge=0,
        le=150,
        description="年龄"
    )
    address: Optional[Address] = Field(
        default=None,
        description="地址"
    )

class UserUpdate(BaseModel):
    """更新用户请求模型"""
    email: Optional[str] = None
    age: Optional[int] = Field(default=None, ge=0, le=150)
    address: Optional[Address] = None

class OrderItem(BaseModel):
    """订单项"""
    product_id: int = Field(ge=1, description="商品ID")
    quantity: int = Field(ge=1, description="数量")
    price: float = Field(ge=0, description="单价")

class OrderCreate(BaseModel):
    """创建订单请求模型"""
    items: List[OrderItem] = Field(min_length=1, description="订单项列表")
    remark: Optional[str] = Field(default=None, max_length=200, description="备注")

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

@app.post("/users", tags=["用户管理"], summary="创建用户")
def create_user(user: UserCreate):
    """
    创建新用户
    
    - **username**: 用户名，3-20个字符
    - **password**: 密码，6-20个字符
    - **email**: 邮箱地址
    - **age**: 年龄（可选）
    - **address**: 地址（可选）
    """
    return {
        "message": "用户创建成功",
        "data": {
            "id": 1,
            "username": user.username,
            "email": user.email,
            "age": user.age,
            "address": user.address
        }
    }

@app.put("/users/{user_id}", tags=["用户管理"], summary="更新用户")
def update_user(
    user_id: int,
    user: UserUpdate,
    notify: bool = Query(default=False, description="是否发送通知")
):
    """更新用户信息"""
    return {
        "message": "更新成功",
        "user_id": user_id,
        "updated_fields": user.model_dump(exclude_none=True),
        "notify": notify
    }

@app.post("/orders", tags=["订单管理"], summary="创建订单")
def create_order(order: OrderCreate):
    """创建新订单"""
    total = sum(item.price * item.quantity for item in order.items)
    return {
        "message": "订单创建成功",
        "order_no": "ORD20231201001",
        "items_count": len(order.items),
        "total": total,
        "remark": order.remark
    }

@app.post("/feedback", tags=["其他"], summary="提交反馈")
def submit_feedback(
    title: str = Body(..., min_length=1, max_length=100),
    content: str = Body(..., min_length=10, max_length=1000),
    contact: Optional[str] = Body(default=None)
):
    """提交用户反馈"""
    return {
        "message": "反馈提交成功",
        "title": title,
        "content": content,
        "contact": contact
    }

📝 小结

本章我们学习了：

  1. ✅ 使用 Pydantic 定义数据模型
  2. ✅ 自动数据验证
  3. ✅ 可选字段和默认值
  4. ✅ 使用 Field 进行字段验证
  5. ✅ 嵌套模型
  6. ✅ 同时使用请求体、路径参数、查询参数

🏃 下一步

学会了接收数据，接下来学习如何控制返回数据的格式！

👉 第7章 - 响应模型

💪 练习题

  1. 创建一个商品模型，包含：name（必填，2-50字符）、price（必填，大于0）、description（可选）、stock（默认0，大于等于0）

  2. 创建一个注册接口 /register，接收用户名、密码、确认密码、邮箱

  3. 创建一个订单模型，包含嵌套的订单项列表

参考答案
from typing import List, Optional
from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()

# 练习1
class Product(BaseModel):
    name: str = Field(min_length=2, max_length=50)
    price: float = Field(gt=0)
    description: Optional[str] = None
    stock: int = Field(default=0, ge=0)

@app.post("/products")
def create_product(product: Product):
    return product

# 练习2
class RegisterRequest(BaseModel):
    username: str = Field(min_length=3, max_length=20)
    password: str = Field(min_length=6)
    confirm_password: str = Field(min_length=6)
    email: str

@app.post("/register")
def register(data: RegisterRequest):
    if data.password != data.confirm_password:
        return {"error": "两次密码不一致"}
    return {"message": "注册成功", "username": data.username}

# 练习3
class OrderItem(BaseModel):
    product_id: int
    quantity: int = Field(ge=1)
    price: float = Field(ge=0)

class Order(BaseModel):
    customer_name: str
    items: List[OrderItem]
    total: float

@app.post("/orders")
def create_order(order: Order):
    return order
最近更新:
Contributors: 王长安