『FastAPI』快速掌握”请求与响应”的基础用法

一、FastAPI请求处理核心机制

FastAPI基于Starlette和Pydantic构建的请求处理系统，通过装饰器模式实现路径操作函数的声明式定义。其核心请求处理流程包含三个关键阶段：

1. 路径匹配阶段：根据请求路径和方法匹配对应的路由处理函数
2. 依赖解析阶段：自动解析路径参数、查询参数和请求体
3. 响应生成阶段：将处理结果序列化为指定格式的响应

from fastapi import FastAPI, Path, Query
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(..., description="物品唯一标识"),
    q: str = Query(None, max_length=50)
):
    return {"item_id": item_id, "q": q}

二、请求参数处理详解

1. 路径参数处理

FastAPI通过类型注解自动转换路径参数类型，支持内置类型和自定义类型：

@app.get("/users/{user_id}/posts/{post_id}")
async def get_post(
    user_id: int,
    post_id: int = Path(..., ge=1, le=1000)
):
    return {"user_id": user_id, "post_id": post_id}

关键特性：

• 自动类型转换（字符串转数字）
• 参数验证（通过Path的ge/le参数）
• 路径参数顺序无关性（通过类型注解匹配）

2. 查询参数处理

查询参数支持多种声明方式：

@app.get("/search/")
async def search_items(
    q: str = None,  # 可选参数
    limit: int = Query(10, le=100),  # 带默认值的必选参数
    sort: str = Query(..., regex="^(asc|desc)$")  # 正则验证
):
    results = {"items": [f"Item {i}" for i in range(limit)]}
    return results

高级用法：

• 参数别名：alias="search_term"
• 弃用参数：deprecated=True
• 多值参数：tags: List[str] = Query(...)

3. 请求体处理

FastAPI通过Pydantic模型实现自动请求体验证：

from pydantic import BaseModel
class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None
@app.post("/items/")
async def create_item(item: Item):
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

验证机制：

• 自动类型转换
• 字段必选/可选控制
• 嵌套模型验证
• 自定义验证器

三、响应处理高级技巧

1. 响应模型控制

通过response_model参数控制返回数据结构：

from fastapi import FastAPI
from pydantic import BaseModel
class OutputItem(BaseModel):
    id: int
    name: str
@app.post("/items/", response_model=OutputItem)
async def create_item(item: Item):
    # 业务逻辑处理
    return OutputItem(id=1, name=item.name)

优势：

• 数据过滤（排除敏感字段）
• 序列化控制
• 文档自动生成

2. 状态码管理

FastAPI支持显式状态码设置：

from fastapi import status
@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    return {"message": "Item created successfully"}

常用状态码：

• 200 OK：默认GET响应
• 201 Created：资源创建
• 400 Bad Request：客户端错误
• 404 Not Found：资源不存在

3. 自定义响应

支持多种响应形式：

from fastapi.responses import JSONResponse, HTMLResponse, StreamingResponse
@app.get("/json")
async def get_json():
    return JSONResponse({"key": "value"})
@app.get("/html")
async def get_html():
    return HTMLResponse("<h1>Hello World</h1>")
@app.get("/stream")
async def stream_data():
    def generate():
        for i in range(10):
            yield f"Data chunk {i}\n"
    return StreamingResponse(generate(), media_type="text/plain")

四、最佳实践与性能优化

1. 参数验证策略

• 使用Field进行详细参数说明：
  ```python
  from pydantic import Field

class Item(BaseModel):
name: str = Field(…, min_length=3, max_length=50)
price: float = Field(…, gt=0)

- 组合验证器：
```python
from pydantic import validator
class Item(BaseModel):
    price: float
    discount: float = 0
    @validator('discount')
    def validate_discount(cls, v, values):
        if 'price' in values and v > values['price']:
            raise ValueError('Discount cannot exceed price')
        return v

2. 响应性能优化

• 使用response_model_exclude_unset减少数据传输：

  @app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)
  async def get_item(item_id: int):
    # 返回可能包含未设置字段的对象
    pass

• 异步响应处理：

  @app.get("/async-data")
  async def get_async_data():
    # 模拟异步数据获取
    await asyncio.sleep(1)
    return {"data": "async result"}

3. 调试与错误处理

自定义异常处理器：

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import HTTPException
app = FastAPI()
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": exc.detail},
    )
@app.get("/error")
async def trigger_error():
    raise HTTPException(status_code=404, detail="Item not found")

五、完整示例：综合应用

from fastapi import FastAPI, Path, Query, HTTPException
from pydantic import BaseModel, Field, validator
from typing import Optional, List
app = FastAPI()
# 请求模型
class Product(BaseModel):
    name: str = Field(..., min_length=2, max_length=100)
    price: float = Field(..., gt=0)
    category: str
    tags: List[str] = []
    @validator('tags')
    def validate_tags(cls, v):
        if len(v) > 10:
            raise ValueError('Maximum 10 tags allowed')
        return v
# 响应模型
class ProductResponse(BaseModel):
    id: int
    name: str
    price: float
    formatted_price: str
    category: str
@app.post("/products/", response_model=ProductResponse, status_code=201)
async def create_product(
    product: Product,
    product_id: int = Path(..., description="Product unique identifier"),
    verbose: bool = Query(False, description="Include detailed information")
):
    if product.price > 1000 and product.category not in ["premium", "luxury"]:
        raise HTTPException(
            status_code=400,
            detail="High price products must belong to premium/luxury categories"
        )
    formatted_price = f"${product.price:.2f}"
    response = ProductResponse(
        id=product_id,
        name=product.name,
        price=product.price,
        formatted_price=formatted_price,
        category=product.category
    )
    if verbose:
        response.tags = product.tags
    return response
# 运行命令：uvicorn main:app --reload

六、总结与进阶建议

FastAPI的请求与响应处理机制通过类型注解和Pydantic模型实现了：

1. 自动参数解析与验证
2. 结构化响应生成
3. 完整的OpenAPI文档生成

进阶学习方向：

• 依赖注入系统
• WebSocket支持
• 背景任务处理
• 中间件开发
• 安全认证集成

建议开发者从基础参数处理入手，逐步掌握响应模型控制，最终实现完整的RESTful API开发。FastAPI的自动文档和类型安全特性可以显著提升开发效率和代码质量。