一、FastAPI请求处理机制详解

1.1 路径参数与类型注解

FastAPI通过Python类型注解实现自动参数解析，路径参数需在路由路径中用花括号{}定义，并在函数参数中声明类型：

from fastapi import FastAPI
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int):  # 类型注解触发自动转换
    return {"item_id": item_id, "type": "integer"}

当访问/items/5时，FastAPI自动将字符串”5”转为整数。支持类型包括int、float、str、bool及自定义类型。

1.2 查询参数处理

查询参数通过函数参数直接声明，支持默认值和可选参数：

@app.get("/search/")
async def search_items(
    query: str,  # 必选参数
    limit: int = 10,  # 可选参数带默认值
    sort: str = None  # 可选参数
):
    results = {"query": query, "limit": limit}
    if sort:
        results["sort"] = sort
    return results

访问/search/?query=test&limit=5&sort=asc时，参数自动绑定到对应变量。

1.3 请求体解析（Pydantic模型）

对于JSON请求体，需定义Pydantic模型实现数据验证：

from pydantic import BaseModel
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.dict()  # 转换为字典
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

发送POST请求时，FastAPI自动验证JSON结构，缺失必填字段会返回422错误。

二、响应处理进阶技巧

2.1 响应模型控制

通过response_model参数精确控制返回字段：

@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
    return {"name": "Test", "price": 10.5, "description": "Test item"}
    # 即使返回字典，也会按Item模型过滤字段

2.2 自定义响应

使用Response对象直接控制响应头和状态码：

from fastapi import Response
@app.get("/download/")
async def download_file():
    content = b"Binary file content"
    headers = {"Content-Disposition": "attachment; filename=test.bin"}
    return Response(content, headers=headers, media_type="application/octet-stream")

2.3 流式响应

处理大文件或实时数据时使用生成器：

@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")

三、生产级实践建议

3.1 路径操作装饰器组合

合理使用多个装饰器处理不同HTTP方法：

@app.get("/items/{item_id}")
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    # 共用逻辑处理
    return {"item_id": item_id, **item.dict()}

3.2 依赖注入系统

使用Depends实现跨路由共享逻辑：

from fastapi import Depends, Header, HTTPException
async def verify_token(x_token: str = Header(...)):
    if x_token != "fake-token":
        raise HTTPException(status_code=403, detail="Invalid token")
    return x_token
@app.get("/secure/")
async def secure_endpoint(token: str = Depends(verify_token)):
    return {"token": token}

3.3 性能优化技巧

对静态文件使用StaticFiles
启用ASGI服务器（如Uvicorn）的异步特性
使用@cache装饰器缓存响应

四、完整示例：订单处理API

from fastapi import FastAPI, Path, Query
from pydantic import BaseModel, EmailStr
from typing import Optional, List
app = FastAPI()
class OrderItem(BaseModel):
    product_id: int
    quantity: int = Query(1, ge=1)
class Order(BaseModel):
    customer_email: EmailStr
    items: List[OrderItem]
    discount: Optional[float] = 0.0
@app.post("/orders/")
async def create_order(order: Order, user_agent: Optional[str] = Header(None)):
    total = sum(item.quantity * 10.0 for item in order.items)  # 假设单价10
    discounted_total = total * (1 - order.discount)
    return {
        "order_id": 1000 + len(app.state.orders) if hasattr(app.state, 'orders') else 1000,
        "total": total,
        "discounted_total": discounted_total,
        "user_agent": user_agent
    }
@app.get("/orders/{order_id}")
async def get_order(
    order_id: int = Path(..., ge=1000),
    include_items: bool = Query(True)
):
    base_order = {"order_id": order_id, "status": "processed"}
    if include_items:
        base_order["items"] = [{"product_id": 1, "quantity": 2}]
    return base_order

五、常见问题解决方案

参数解析失败：检查类型注解是否正确，使用str接收无法转换的值
CORS问题：添加@app.middleware("http")或使用fastapi.middleware.cors.CORSMiddleware
性能瓶颈：用anyio进行并发处理，避免同步IO操作
版本控制：在路由前缀中添加版本号，如@app.get("/v1/items/")

通过系统掌握这些核心机制，开发者能够快速构建出符合企业级标准的RESTful API。建议结合FastAPI官方文档和Pydantic模型验证文档进行深入学习，同时利用Swagger UI（访问/docs）进行交互式测试。

「FastAPI实战」请求与响应核心机制全解析