FastAPIrequest-body-handling and response model

📂 Stage: Stage 1 - Rapid Foundation Building (Basics) 🔗 Related chapters: path-query-parameters · response-handling-status-codes · pydantic-guide

Table of contents

request-body-handling overview

The request body is the heart of Web API receiving client data. FastAPI and Pydantic 2.x have joined forces to provide you with a three-in-one experience of "automatic verification + automatic documentation + type conversion", completely saying goodbye to the repetitive work of manually parsing JSON and handwriting verification logic.

Core processing flow

  1. Use Pydantic model to define data contract: field type, whether it is required, and constraint rules are clear at a glance
  2. FastAPI automatically parses and verifies: extract data from HTTP requests, convert it into the correct Python type and verify it
  3. Friendly error feedback: When the verification fails, a structured error message is returned to facilitate quick location of the client.
  4. OpenAPI document automatically generated: constraints in the model are directly mapped to examples and instructions in the interactive document

Pydantic request model

Basic model and field constraints

The most common approach is to inheritBaseModel, and then cooperateFieldAdd detailed constraints and build the request body like building blocks.

from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime
import re

app = FastAPI()

class ProductCreate(BaseModel):
    """产品创建请求模型"""
    # 必填字段，附带最小/最大长度约束
    name: str = Field(..., min_length=3, max_length=100, description="产品名称")
    # 可选字段，可以设默认值 None 或具体值
    description: Optional[str] = Field(None, max_length=500, description="产品描述")
    # 数值约束：大于0，小于等于10000
    price: float = Field(..., gt=0, le=10000, description="产品价格")
    # 正则约束：字母数字下划线开头，后续可包含字母数字下划线
    category: str = Field(..., regex=r"^[a-zA-Z_][a-zA-Z0-9_]*$", description="产品分类")
    # 列表约束：最多10个标签
    tags: List[str] = Field(default=[], max_items=10, description="产品标签")
    # 使用工厂函数生成默认时间，避免固定值
    created_at: datetime = Field(default_factory=datetime.utcnow)
    is_active: bool = True

@app.post("/products/")
def create_product(product: ProductCreate):
    """
    创建产品
    自动完成JSON解析、类型转换、数据验证
    """
    # Pydantic 2.x 推荐使用 model_dump() 替代 dict()
    return product.model_dump()

Custom business verification

When the built-in constraints are not enough, you can use@validatorProcess individual fields, or use@root_validatorPerform multi-field linkage to implement complex business logic.

from pydantic import validator, root_validator

class OrderItem(BaseModel):
    product_id: int
    quantity: int = Field(..., gt=0, le=1000)
    unit_price: float = Field(..., gt=0)

class OrderCreate(BaseModel):
    customer_id: int
    items: List[OrderItem] = Field(..., min_items=1)
    order_date: datetime
    delivery_date: datetime
    discount_code: Optional[str] = None

    @validator('discount_code')
    def validate_discount(cls, v):
        """单字段验证：折扣码格式检查"""
        if v and not re.match(r"^[A-Z0-9]{6,10}$", v):
            raise ValueError('折扣码格式无效（需要6-10位大写字母和数字）')
        return v.upper() if v else v

    @root_validator
    def validate_order_logic(cls, values):
        """多字段联动验证：业务规则"""
        order_date = values.get('order_date')
        delivery_date = values.get('delivery_date')
        discount_code = values.get('discount_code')

        # 配送时间必须严格晚于下单时间
        if order_date and delivery_date and delivery_date <= order_date:
            raise ValueError('配送时间必须晚于下单时间')
        # 订单总金额未满100元时，不能使用折扣码
        if discount_code and values.get('items'):
            total = sum(i.quantity * i.unit_price for i in values['items'])
            if total < 100:
                raise ValueError('小额订单（<100元）不能使用折扣码')
        return values

Response model definition

useresponse_modelThe fields returned to the client can be strictly controlled, which not only avoids leaking sensitive information such as passwords, but also makes the response structure more unified.

from pydantic import EmailStr     # Pydantic 2.x 内置的邮箱验证
from fastapi import Path
from typing import TypeVar, Generic, Optional
from datetime import datetime, timezone

# 去除敏感字段后的用户响应模型
class UserResponse(BaseModel):
    id: int
    username: str
    email: EmailStr
    created_at: datetime
    is_active: bool
    # 注意：password字段已被排除在外，不会出现在响应中

# 泛型API响应模型，统一整体的返回格式
T = TypeVar('T')
class ApiResponse(BaseModel, Generic[T]):
    success: bool = True
    message: str = "操作成功"
    data: Optional[T] = None
    timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

# 模拟用户数据库
fake_db = {
    1: {
        "id": 1,
        "username": "johndoe",
        "email": "john@example.com",
        "password": "secret",
        "created_at": datetime.now(timezone.utc),
        "is_active": True
    }
}

@app.get("/users/{user_id}", response_model=ApiResponse[UserResponse])
def get_user(user_id: int = Path(..., gt=0)):
    """获取用户，返回统一格式的响应"""
    user_data = fake_db.get(user_id)
    if not user_data:
        return ApiResponse[UserResponse](success=False, message="用户不存在", data=None)
    return ApiResponse[UserResponse](data=user_data)

File upload and form processing

Basic file upload

FastAPIUploadFileDesigned specifically for large files, using streaming reading with extremely low memory usage. You can easily process single or multiple files.

from fastapi import File, UploadFile
from typing import List
import aiofiles
from pathlib import Path

UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

@app.post("/uploadfile/")
async def upload_file(file: UploadFile = File(...)):
    """上传单个文件并异步保存"""
    # 生成唯一文件名，防止覆盖
    unique_name = f"{file.filename.split('.')[0]}_{hash(file.filename)}.{file.filename.split('.')[-1]}"
    save_path = UPLOAD_DIR / unique_name

    # 分块流式保存，每次读取 1MB
    async with aiofiles.open(save_path, 'wb') as f:
        while content := await file.read(1024 * 1024):
            await f.write(content)

    return {"filename": unique_name, "size": save_path.stat().st_size}

@app.post("/uploadfiles/")
async def upload_files(files: List[UploadFile] = File(...)):
    """一次上传多个文件"""
    results = []
    for file in files:
        size = len(await file.read())
        results.append({
            "filename": file.filename,
            "content_type": file.content_type,
            "size": size
        })
    return results

Form data processing

Process normal HTML forms, ormultipart/form-dataFor non-JSON data in the format, just add on the parameterForm(...)。

from fastapi import Form

@app.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)):
    """基础表单登录"""
    return {"username": username, "message": "登录请求接收成功"}

Nested models and complex data

Flat data structures are rarely encountered in real business. Nested models can help you clearly describe complex relationships such as orders, customers, addresses, etc.

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

class CustomerResponse(BaseModel):
    id: int
    name: str
    email: EmailStr
    address: Optional[Address] = None

class OrderItemResponse(BaseModel):
    product_name: str
    quantity: int
    total_price: float

class OrderResponse(BaseModel):
    order_id: str
    customer: CustomerResponse
    items: List[OrderItemResponse]
    total_amount: float
    status: str

# 模拟一个包含嵌套数据的完整订单
sample_order = OrderResponse(
    order_id="ORD2024010001",
    customer=CustomerResponse(
        id=1,
        name="John Doe",
        email="john@example.com",
        address=Address(street="123 Main St", city="New York", country="USA", postal_code="10001")
    ),
    items=[
        OrderItemResponse(product_name="Laptop", quantity=1, total_price=999.99),
        OrderItemResponse(product_name="Mouse", quantity=2, total_price=59.98)
    ],
    total_amount=1059.97,
    status="processing"
)

Error handling and performance optimization

Request body verification error handling

by customizingRequestValidationErrorThe exception-handler can convert boring default errors into a clearly structured and caller-friendly format.

from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    """统一结构化验证错误响应"""
    errors = []
    for err in exc.errors():
        errors.append({
            "field": ".".join(str(loc) for loc in err["loc"][1:]),  # 去除 body 前缀
            "message": err["msg"],
            "type": err["type"]
        })
    return JSONResponse(
        status_code=422,
        content=ApiResponse[None](
            success=False,
            message="请求数据验证失败",
            data={"errors": errors}
        ).model_dump()
    )

Basic performance optimization suggestions

  • Make good use of async: Use all IO-intensive operations such as file reading and writing, database queries, external API calls, etc.async/awaitAvoid blocking.
  • Simplified Validation: Do not abuse extremely complex regular expressions in the request model. For problems that can be solved with built-in type constraints, use built-in methods first.
  • Large file chunked upload: PassedUploadFile.read(size)Process in chunks to avoid reading everything into memory at once.
  • Control Response Volume: used appropriatelyresponse_model_exclude_unsetorresponse_model_exclude_defaults, only return truly meaningful fields.

Practical application cases

The following is a complete Create Order API, which integrates various techniques introduced previously: request model verification, response model control, background task processing, etc.

from fastapi import BackgroundTasks
import uuid
import asyncio

# 模拟数据库操作
async def save_order_to_db(order_data: dict):
    print(f"保存订单到数据库: {order_data}")
    await asyncio.sleep(0.1)

# 模拟邮件通知
async def send_order_confirmation(order_id: str, email: str):
    print(f"发送订单确认邮件到 {email}，订单号 {order_id}")
    await asyncio.sleep(0.2)

@app.post("/orders/", response_model=ApiResponse[OrderResponse])
async def create_order_api(
    order: OrderCreate,
    background_tasks: BackgroundTasks
):
    """完整的创建订单 API"""
    # 生成唯一的订单号
    order_id = f"ORD-{uuid.uuid4().hex[:8].upper()}"
    # 计算订单总金额
    total_amount = sum(item.quantity * item.unit_price for item in order.items)
    # 组装响应数据（示例中写死部分客户信息，生产环境应查询数据库）
    order_response = OrderResponse(
        order_id=order_id,
        customer=CustomerResponse(
            id=order.customer_id,
            name="John Doe",
            email="john@example.com",
            address=Address(street="123 Main St", city="New York", country="USA", postal_code="10001")
        ),
        items=[
            OrderItemResponse(
                product_name=f"Product {item.product_id}",
                quantity=item.quantity,
                total_price=item.quantity * item.unit_price
            )
            for item in order.items
        ],
        total_amount=total_amount,
        status="processing"
    )
    # 将耗时操作放入后台任务，避免阻塞接口返回
    background_tasks.add_task(save_order_to_db, order.model_dump())
    background_tasks.add_task(send_order_confirmation, order_id, "john@example.com")
    return ApiResponse[OrderResponse](data=order_response, message="订单创建成功")

:::tip request-body-handling best practices

  1. Clearly distinguish between "request model" and "response model" to avoid inadvertently returning sensitive information
  2. Prefer using the built-in validators of Pydantic 2.x (such asEmailStr、PaymentCardNumber), reduce custom regular
  3. Multi-field linkage verification is handed over@root_validator, maintain the integrity of business logic
  4. Be sure to use it when uploading large files.UploadFile.read(size)Read in chunks to prevent memory overflow :::

Summarize

FastAPI's request body and response model is one of its most dazzling capabilities. With the power of Pydantic 2.x, you can get a secure, standardized and efficient data transfer experience with only a small amount of code. By skillfully applying these techniques, you can quickly build an enterprise-level Web API.