FastAPI 请求和响应

概述

HTTP请求和响应是Web API的核心。FastAPI提供了强大的请求处理和响应生成机制，支持自动验证、序列化、文档生成等功能。本章将深入探讨如何处理各种类型的请求数据和定制响应格式。

📥 请求体处理

Pydantic模型请求体

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field, validator
from typing import Optional, List
from datetime import datetime
from enum import Enum

app = FastAPI()

class Priority(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    URGENT = "urgent"

class TaskCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200, description="任务标题")
    description: Optional[str] = Field(None, max_length=1000, description="任务描述")
    priority: Priority = Field(Priority.MEDIUM, description="任务优先级")
    due_date: Optional[datetime] = Field(None, description="截止日期")
    tags: List[str] = Field(default_factory=list, description="任务标签")
    estimated_hours: Optional[float] = Field(None, ge=0.1, le=1000, description="预估工时")

    @validator('tags')
    def validate_tags(cls, v):
        if len(v) > 10:
            raise ValueError('标签数量不能超过10个')
        return [tag.strip().lower() for tag in v if tag.strip()]

    @validator('due_date')
    def validate_due_date(cls, v):
        if v and v < datetime.now():
            raise ValueError('截止日期不能是过去的时间')
        return v

class TaskResponse(BaseModel):
    id: int
    title: str
    description: Optional[str]
    priority: Priority
    due_date: Optional[datetime]
    tags: List[str]
    estimated_hours: Optional[float]
    created_at: datetime
    updated_at: datetime
    is_completed: bool = False

    class Config:
        from_attributes = True

# 创建任务
@app.post("/tasks/", response_model=TaskResponse, status_code=status.HTTP_201_CREATED)
async def create_task(task: TaskCreate):
    # 模拟创建任务
    task_data = task.dict()
    task_data.update({
        "id": 1,
        "created_at": datetime.now(),
        "updated_at": datetime.now(),
        "is_completed": False
    })
    
    return TaskResponse(**task_data)

嵌套模型

class Address(BaseModel):
    street: str = Field(..., min_length=1, max_length=200)
    city: str = Field(..., min_length=1, max_length=100)
    state: str = Field(..., min_length=2, max_length=50)
    postal_code: str = Field(..., regex=r"^\d{5}(-\d{4})?$")
    country: str = Field(default="US", max_length=50)

class Contact(BaseModel):
    email: str = Field(..., regex=r"^[^@]+@[^@]+\.[^@]+$")
    phone: Optional[str] = Field(None, regex=r"^\+?1?\d{9,15}$")

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50, regex="^[a-zA-Z0-9_]+$")
    full_name: str = Field(..., min_length=1, max_length=100)
    address: Address
    contact: Contact
    preferences: Optional[dict] = Field(default_factory=dict)

    @validator('preferences')
    def validate_preferences(cls, v):
        allowed_keys = {'theme', 'language', 'notifications', 'timezone'}
        if not all(key in allowed_keys for key in v.keys()):
            raise ValueError(f'preferences只能包含: {allowed_keys}')
        return v

@app.post("/users/", response_model=dict)
async def create_user(user: UserCreate):
    return {
        "message": "用户创建成功",
        "user": user.dict(),
        "summary": {
            "username": user.username,
            "location": f"{user.address.city}, {user.address.state}",
            "contact_email": user.contact.email
        }
    }

多个请求体参数

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

class User(BaseModel):
    username: str
    full_name: str

class Metadata(BaseModel):
    source: str = "api"
    timestamp: datetime = Field(default_factory=datetime.now)

@app.post("/items/{item_id}")
async def update_item_with_user(
    item_id: int,
    item: Item,
    user: User,
    metadata: Metadata,
    importance: int = Field(..., ge=1, le=5, description="重要程度 1-5")
):
    return {
        "item_id": item_id,
        "item": item.dict(),
        "user": user.dict(),
        "metadata": metadata.dict(),
        "importance": importance,
        "operation": "update_with_context"
    }

📤 响应模型

基础响应模型

from typing import Union

class ErrorResponse(BaseModel):
    error: str
    message: str
    details: Optional[dict] = None

class SuccessResponse(BaseModel):
    success: bool = True
    message: str
    data: Optional[dict] = None

class PaginatedResponse(BaseModel):
    items: List[dict]
    total: int
    page: int
    per_page: int
    pages: int

    @validator('page')
    def validate_page(cls, v, values):
        if 'pages' in values and v > values['pages']:
            raise ValueError('页码不能超过总页数')
        return v

# 使用Union类型的响应
@app.get("/items/{item_id}", response_model=Union[TaskResponse, ErrorResponse])
async def get_item(item_id: int):
    if item_id <= 0:
        return ErrorResponse(
            error="invalid_id",
            message="物品ID必须大于0",
            details={"provided_id": item_id}
        )
    
    # 模拟获取物品
    if item_id == 999:
        return ErrorResponse(
            error="not_found",
            message="未找到指定的物品"
        )
    
    return TaskResponse(
        id=item_id,
        title="示例任务",
        description="这是一个示例任务",
        priority=Priority.MEDIUM,
        created_at=datetime.now(),
        updated_at=datetime.now()
    )

响应模型继承

class BaseResponse(BaseModel):
    timestamp: datetime = Field(default_factory=datetime.now)
    api_version: str = "1.0"

class UserResponse(BaseResponse):
    id: int
    username: str
    email: str
    is_active: bool

class UserListResponse(BaseResponse):
    users: List[UserResponse]
    total_count: int

class UserDetailResponse(UserResponse):
    full_name: str
    created_at: datetime
    last_login: Optional[datetime]
    profile: Optional[dict]

@app.get("/users/", response_model=UserListResponse)
async def list_users():
    return UserListResponse(
        users=[
            UserResponse(id=1, username="alice", email="alice@example.com", is_active=True),
            UserResponse(id=2, username="bob", email="bob@example.com", is_active=False)
        ],
        total_count=2
    )

@app.get("/users/{user_id}", response_model=UserDetailResponse)
async def get_user_detail(user_id: int):
    return UserDetailResponse(
        id=user_id,
        username="alice",
        email="alice@example.com",
        is_active=True,
        full_name="Alice Johnson",
        created_at=datetime.now(),
        last_login=datetime.now(),
        profile={"bio": "Software developer", "location": "San Francisco"}
    )

🎭 响应定制

多种响应格式

from fastapi import Response
from fastapi.responses import JSONResponse, HTMLResponse, PlainTextResponse

@app.get("/data/{format}")
async def get_data_in_format(format: str, item_id: int = 1):
    data = {"id": item_id, "name": "Sample Item", "value": 42}
    
    if format == "json":
        return JSONResponse(content=data)
    elif format == "html":
        html_content = f"""
        <html>
            <body>
                <h1>Item {data['id']}</h1>
                <p>Name: {data['name']}</p>
                <p>Value: {data['value']}</p>
            </body>
        </html>
        """
        return HTMLResponse(content=html_content)
    elif format == "text":
        text_content = f"ID: {data['id']}, Name: {data['name']}, Value: {data['value']}"
        return PlainTextResponse(content=text_content)
    else:
        raise HTTPException(
            status_code=400,
            detail="不支持的格式，支持的格式: json, html, text"
        )

条件响应

from fastapi.responses import RedirectResponse

@app.get("/items/{item_id}")
async def get_item_conditional(
    item_id: int,
    format: str = "json",
    redirect_if_inactive: bool = False
):
    # 模拟物品数据
    item = {
        "id": item_id,
        "name": "Sample Item",
        "is_active": item_id % 2 == 0,  # 偶数ID为活跃
        "created_at": datetime.now().isoformat()
    }
    
    # 条件重定向
    if not item["is_active"] and redirect_if_inactive:
        return RedirectResponse(
            url=f"/items/active?suggested_id={item_id + 1}",
            status_code=status.HTTP_307_TEMPORARY_REDIRECT
        )
    
    # 条件响应格式
    if format == "summary":
        return {"id": item["id"], "name": item["name"]}
    elif format == "full":
        return item
    else:
        # 默认格式
        return {"id": item["id"], "active": item["is_active"]}

📋 状态码管理

自定义状态码

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item_with_status(item: Item):
    return {"message": "物品创建成功", "item": item.dict()}

@app.put("/items/{item_id}")
async def update_item_with_conditional_status(item_id: int, item: Item):
    # 模拟检查物品是否存在
    item_exists = item_id <= 100
    
    if not item_exists:
        # 创建新物品
        return JSONResponse(
            status_code=status.HTTP_201_CREATED,
            content={
                "message": "物品不存在，已创建新物品",
                "item_id": item_id,
                "item": item.dict()
            }
        )
    else:
        # 更新现有物品
        return JSONResponse(
            status_code=status.HTTP_200_OK,
            content={
                "message": "物品更新成功",
                "item_id": item_id,
                "item": item.dict()
            }
        )

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
    # 删除成功，返回204 No Content
    return Response(status_code=status.HTTP_204_NO_CONTENT)

错误响应处理

class ValidationErrorResponse(BaseModel):
    error_type: str = "validation_error"
    message: str
    field_errors: List[dict]

class NotFoundErrorResponse(BaseModel):
    error_type: str = "not_found"
    message: str
    resource: str

@app.get("/items/{item_id}")
async def get_item_with_error_handling(item_id: int):
    if item_id <= 0:
        return JSONResponse(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            content=ValidationErrorResponse(
                message="物品ID验证失败",
                field_errors=[{
                    "field": "item_id",
                    "error": "必须大于0",
                    "provided_value": item_id
                }]
            ).dict()
        )
    
    if item_id == 404:
        return JSONResponse(
            status_code=status.HTTP_404_NOT_FOUND,
            content=NotFoundErrorResponse(
                message="物品未找到",
                resource="item"
            ).dict()
        )
    
    return {"item_id": item_id, "name": "Sample Item"}

🔧 请求头和响应头

处理请求头

from fastapi import Header, Request

@app.get("/info/")
async def get_request_info(
    request: Request,
    user_agent: Optional[str] = Header(None),
    accept_language: Optional[str] = Header(None, alias="accept-language"),
    x_api_key: Optional[str] = Header(None, alias="x-api-key"),
    authorization: Optional[str] = Header(None)
):
    return {
        "client_ip": request.client.host,
        "user_agent": user_agent,
        "accept_language": accept_language,
        "has_api_key": x_api_key is not None,
        "has_auth": authorization is not None,
        "all_headers": dict(request.headers)
    }

@app.post("/upload/")
async def upload_file(
    content_type: str = Header(...),
    content_length: int = Header(...),
    x_file_name: Optional[str] = Header(None, alias="x-file-name")
):
    if content_length > 10 * 1024 * 1024:  # 10MB
        raise HTTPException(
            status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
            detail="文件大小不能超过10MB"
        )
    
    return {
        "content_type": content_type,
        "content_length": content_length,
        "file_name": x_file_name,
        "status": "ready_to_process"
    }

设置响应头

from fastapi import Response

@app.get("/download/{file_id}")
async def download_file(file_id: int, response: Response):
    # 设置下载响应头
    response.headers["Content-Disposition"] = f"attachment; filename=file_{file_id}.txt"
    response.headers["Content-Type"] = "text/plain"
    response.headers["X-Download-ID"] = str(file_id)
    
    return {"content": f"这是文件 {file_id} 的内容"}

@app.get("/api/data/")
async def get_api_data():
    content = {"data": "sample", "timestamp": datetime.now().isoformat()}
    
    headers = {
        "X-API-Version": "1.0",
        "X-Rate-Limit": "100",
        "X-Rate-Remaining": "95",
        "Cache-Control": "public, max-age=300"
    }
    
    return JSONResponse(content=content, headers=headers)

🍪 Cookie处理

读取和设置Cookie

from fastapi import Cookie

@app.get("/profile/")
async def get_profile(
    session_id: Optional[str] = Cookie(None),
    user_preferences: Optional[str] = Cookie(None, alias="user-preferences")
):
    return {
        "has_session": session_id is not None,
        "session_id": session_id,
        "preferences": user_preferences,
        "message": "获取用户配置"
    }

@app.post("/login/")
async def login(username: str, password: str, response: Response):
    # 模拟登录验证
    if username == "admin" and password == "secret":
        # 设置会话Cookie
        response.set_cookie(
            key="session_id",
            value="abc123xyz",
            max_age=3600,  # 1小时
            httponly=True,
            secure=True,  # 仅HTTPS
            samesite="strict"
        )
        
        # 设置偏好Cookie
        response.set_cookie(
            key="user-preferences",
            value="theme=dark;lang=zh",
            max_age=86400 * 30  # 30天
        )
        
        return {"message": "登录成功", "username": username}
    else:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误"
        )

@app.post("/logout/")
async def logout(response: Response):
    # 删除Cookie
    response.delete_cookie(key="session_id")
    response.delete_cookie(key="user-preferences")
    
    return {"message": "已退出登录"}

📊 数据验证和转换

自定义验证器

from pydantic import BaseModel, validator, root_validator
import re

class ProductCreate(BaseModel):
    name: str = Field(..., min_length=2, max_length=100)
    sku: str = Field(..., description="产品SKU")
    price: float = Field(..., gt=0, description="产品价格")
    category: str
    specifications: dict = Field(default_factory=dict)
    
    @validator('sku')
    def validate_sku(cls, v):
        # SKU格式验证: 字母开头，后跟数字和字母
        if not re.match(r'^[A-Z]{2,3}\d{3,6}$', v.upper()):
            raise ValueError('SKU格式错误，应为2-3个字母开头加3-6个数字')
        return v.upper()
    
    @validator('category')
    def validate_category(cls, v):
        allowed_categories = ['electronics', 'clothing', 'books', 'home', 'sports']
        if v.lower() not in allowed_categories:
            raise ValueError(f'分类必须是: {", ".join(allowed_categories)}')
        return v.lower()
    
    @root_validator
    def validate_product(cls, values):
        # 跨字段验证
        category = values.get('category')
        price = values.get('price')
        
        if category == 'electronics' and price < 10:
            raise ValueError('电子产品价格不能低于10元')
        
        if category == 'books' and price > 1000:
            raise ValueError('书籍价格不能超过1000元')
        
        return values

@app.post("/products/", response_model=dict)
async def create_product(product: ProductCreate):
    return {
        "message": "产品创建成功",
        "product": product.dict(),
        "generated_id": hash(product.sku) % 10000
    }

数据转换和格式化

class EventCreate(BaseModel):
    title: str
    start_time: datetime
    duration_minutes: int = Field(..., ge=15, le=480)  # 15分钟到8小时
    location: str
    attendees: List[str] = Field(default_factory=list)
    
    @validator('title')
    def clean_title(cls, v):
        # 清理和格式化标题
        return ' '.join(v.strip().split())
    
    @validator('attendees', pre=True)
    def parse_attendees(cls, v):
        # 处理逗号分隔的字符串
        if isinstance(v, str):
            return [email.strip() for email in v.split(',') if email.strip()]
        return v
    
    @validator('attendees')
    def validate_emails(cls, v):
        email_pattern = r'^[^@]+@[^@]+\.[^@]+$'
        for email in v:
            if not re.match(email_pattern, email):
                raise ValueError(f'无效的邮箱地址: {email}')
        return v

@app.post("/events/")
async def create_event(event: EventCreate):
    # 计算结束时间
    end_time = event.start_time + timedelta(minutes=event.duration_minutes)
    
    return {
        "event": event.dict(),
        "computed_fields": {
            "end_time": end_time.isoformat(),
            "duration_hours": event.duration_minutes / 60,
            "attendee_count": len(event.attendees)
        }
    }

总结

本章深入介绍了FastAPI的请求和响应处理：

• ✅ 请求体处理：Pydantic模型、嵌套模型、多参数
• ✅ 响应模型：基础响应、模型继承、条件响应
• ✅ 响应定制：多种格式、状态码管理
• ✅ HTTP头处理：请求头读取、响应头设置
• ✅ Cookie管理：读取、设置、删除Cookie
• ✅ 数据验证：自定义验证器、数据转换

这些功能使FastAPI能够处理复杂的API需求，提供类型安全和自动文档生成的同时，保持代码的简洁性和可维护性。

请求响应设计建议

• 使用Pydantic模型确保数据类型安全
• 提供清晰的错误响应格式
• 合理使用HTTP状态码
• 添加适当的响应头信息
• 考虑API的向后兼容性
• 编写详细的模型文档

在下一章中，我们将学习FastAPI的表单处理和文件上传功能。