请求体（Request Body）处理

一、什么是请求体

请求体（Request Body）是 HTTP POST/PUT/PATCH 请求中发送的数据，通常用于创建或更新资源。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

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

@app.post("/items/")
async def create_item(item: Item):
    # item 是请求体，自动解析和验证
    return item

请求示例：

POST /items/
Content-Type: application/json

{
    "name": "iPhone 15",
    "price": 5999.00
}

二、基本请求体

定义请求模型

from pydantic import BaseModel, Field

class UserCreate(BaseModel):
    username: str = Field(min_length=3, max_length=20)
    email: str
    password: str = Field(min_length=8)
    age: int | None = None

@app.post("/users/")
async def create_user(user: UserCreate):
    return {
        "message": "User created",
        "user": user.model_dump()
    }

可选字段

class Item(BaseModel):
    name: str
    description: str | None = None  # 可选
    price: float
    tax: float | None = None        # 可选

@app.post("/items/")
async def create_item(item: Item):
    item_dict = item.model_dump()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

三、嵌套模型

模型嵌套

class Image(BaseModel):
    url: str
    name: str

class Item(BaseModel):
    name: str
    price: float
    image: Image | None = None  # 嵌套模型

@app.post("/items/")
async def create_item(item: Item):
    return item

请求示例：

{
    "name": "iPhone 15",
    "price": 5999.00,
    "image": {
        "url": "https://example.com/iphone.jpg",
        "name": "iPhone 15 图片"
    }
}

嵌套列表

class Image(BaseModel):
    url: str
    name: str

class Item(BaseModel):
    name: str
    price: float
    images: list[Image] = []  # 图片列表

@app.post("/items/")
async def create_item(item: Item):
    return item

请求示例：

{
    "name": "iPhone 15",
    "price": 5999.00,
    "images": [
        {"url": "https://example.com/1.jpg", "name": "正面"},
        {"url": "https://example.com/2.jpg", "name": "背面"}
    ]
}

四、单字段验证

使用 Field 验证

from pydantic import BaseModel, Field

class Product(BaseModel):
    name: str = Field(
        min_length=1,
        max_length=100,
        description="产品名称"
    )
    price: float = Field(
        gt=0,
        description="产品价格，必须大于0"
    )
    quantity: int = Field(
        ge=0,
        le=1000,
        default=0,
        description="库存数量，0-1000"
    )
    category: str = Field(
        pattern=r"^(electronics|clothing|food)$",
        description="产品类别"
    )

自定义验证器

from pydantic import BaseModel, field_validator

class User(BaseModel):
    username: str
    password: str
    confirm_password: str
    
    @field_validator("password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError("密码至少8位")
        if not any(c.isupper() for c in v):
            raise ValueError("密码必须包含大写字母")
        if not any(c.isdigit() for c in v):
            raise ValueError("密码必须包含数字")
        return v
    
    @field_validator("confirm_password")
    @classmethod
    def passwords_match(cls, v: str, info) -> str:
        if "password" in info.data and v != info.data["password"]:
            raise ValueError("两次输入的密码不一致")
        return v

五、示例配置

model_config 配置示例

from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(min_length=1)
    price: float = Field(gt=0)
    
    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "name": "iPhone 15",
                    "price": 5999.00
                },
                {
                    "name": "MacBook Pro",
                    "price": 14999.00
                }
            ]
        }
    }

Field 配置示例

from pydantic import BaseModel, Field

class User(BaseModel):
    id: int = Field(
        examples=[1, 2, 3],
        description="用户唯一标识"
    )
    name: str = Field(
        examples=["张三", "李四"],
        description="用户姓名"
    )
    email: str = Field(
        examples=["user@example.com"],
        pattern=r"^[\w.-]+@[\w.-]+\.\w+$"
    )

六、多请求体

多个请求体参数

from fastapi import Body

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Item,
    user: User
):
    return {
        "item_id": item_id,
        "item": item,
        "user": user
    }

请求示例：

{
    "item": {
        "name": "iPhone",
        "price": 5999
    },
    "user": {
        "username": "zhangsan",
        "email": "test@example.com"
    }
}

单字段作为请求体

from fastapi import Body

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

七、文件上传

单文件上传

from fastapi import FastAPI, File, UploadFile

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": len(await file.read())
    }

多文件上传

@app.post("/uploads/")
async def upload_files(files: list[UploadFile]):
    return [
        {"filename": f.filename, "size": len(await f.read())}
        for f in files
    ]

文件 + 表单数据

from fastapi import Form

@app.post("/upload-with-info/")
async def upload_with_info(
    file: UploadFile = File(...),
    title: str = Form(...),
    description: str | None = Form(None)
):
    return {
        "title": title,
        "description": description,
        "filename": file.filename
    }

八、请求体 + 路径/查询参数

混合使用

@app.post("/users/{user_id}/items/")
async def create_user_item(
    user_id: int,              # 路径参数
    item: Item,                # 请求体
    q: str | None = None       # 查询参数
):
    return {
        "user_id": user_id,
        "item": item,
        "q": q
    }

九、完整示例

from fastapi import FastAPI, Body, File, UploadFile
from pydantic import BaseModel, Field, field_validator
from typing import Annotated

app = FastAPI()

class Image(BaseModel):
    url: str = Field(pattern=r"^https?://")
    name: str

class Item(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    description: str | None = Field(default=None, max_length=500)
    price: float = Field(gt=0)
    tax: float | None = Field(default=None, ge=0)
    tags: list[str] = []
    image: Image | None = None
    
    model_config = {
        "json_schema_extra": {
            "examples": [{
                "name": "iPhone 15",
                "description": "最新款 iPhone",
                "price": 5999.00,
                "tax": 599.90,
                "tags": ["手机", "苹果"],
                "image": {
                    "url": "https://example.com/iphone.jpg",
                    "name": "iPhone 15"
                }
            }]
        }
    }

@app.post("/items/", response_model=Item)
async def create_item(item: Annotated[Item, Body(embed=True)]):
    return item

@app.post("/upload-image/")
async def upload_image(
    image: UploadFile = File(..., description="产品图片"),
    caption: str | None = Form(None)
):
    content = await image.read()
    return {
        "filename": image.filename,
        "size": len(content),
        "caption": caption
    }

十、小结

功能用法
基本请求体item: Item
嵌套模型字段类型为另一个模型
嵌套列表list[Model]
字段验证Field(gt=0, max_length=100)
自定义验证@field_validator
示例配置model_config = {"json_schema_extra": {...}}
单文件上传file: UploadFile = File(...)
多文件上传files: list[UploadFile]