FastAPI请求与响应实战指南：从基础到进阶

一、FastAPI请求处理核心机制

FastAPI的请求处理基于Python类型注解实现自动参数解析，其核心优势在于无需手动编写参数提取逻辑。当定义路由函数时，FastAPI会通过函数签名自动识别以下参数类型：

1. 路径参数：使用花括号{}在路径中定义，如/items/{item_id}，对应函数参数需标注类型
2. 查询参数：通过?key=value形式传递，函数参数默认即为查询参数
3. 请求体：使用Pydantic模型或字典接收JSON数据
4. 请求头/Cookie：通过Header()和Cookie()特殊函数标注

from fastapi import FastAPI, Query, Path, Header
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
    name: str
    price: float
@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(..., ge=1),  # 路径参数+验证
    q: str = None,                   # 可选查询参数
    limit: int = Query(100, le=500), # 带默认值和约束的查询参数
    x_token: str = Header(...),      # 必需请求头
):
    return {"item_id": item_id, "q": q}

二、响应封装与状态码管理

FastAPI提供三种响应处理方式，开发者可根据场景选择最适合的方案：

1. 自动JSON响应（推荐）

默认情况下，FastAPI会自动将函数返回值序列化为JSON，并设置Content-Type: application/json。对于复杂响应，建议使用Pydantic模型保证类型安全：

from fastapi import FastAPI, Response
from fastapi.responses import JSONResponse, HTMLResponse
@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    # 业务逻辑处理
    return item  # 自动转换为JSON

2. 显式响应对象

当需要精确控制响应头或状态码时，可使用响应类：

@app.get("/custom")
async def custom_response():
    content = {"message": "Custom response"}
    headers = {"X-Custom-Header": "value"}
    return JSONResponse(
        content=content,
        headers=headers,
        status_code=201
    )

3. 状态码最佳实践

FastAPI支持标准HTTP状态码，推荐使用枚举值提高代码可读性：

from fastapi import status
@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    # 删除逻辑
    return Response(status_code=status.HTTP_204_NO_CONTENT)

三、进阶请求处理技巧

1. 多参数组合处理

实际开发中常需同时处理多种参数类型，FastAPI支持参数组合验证：

from typing import Optional
@app.get("/search")
async def search_items(
    query: str,
    skip: int = 0,
    limit: int = Query(10, lte=100),
    sort: Optional[List[str]] = Query(None)
):
    # 处理逻辑
    return {"results": []}

2. 文件上传处理

FastAPI内置对文件上传的支持，可处理单个或多个文件：

from fastapi import UploadFile, File
@app.post("/upload/")
async def upload_file(
    file: UploadFile = File(...),
    description: str = Form(...)
):
    contents = await file.read()
    return {"filename": file.filename, "size": len(contents)}

3. WebSocket请求处理

对于实时通信场景，FastAPI提供WebSocket支持：

from fastapi import WebSocket
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Echo: {data}")

四、性能优化建议

1. 异步处理：对I/O密集型操作使用async/await
2. 响应缓存：对静态内容使用CacheControl中间件
3. 数据验证：充分利用Pydantic模型的验证功能
4. 依赖注入：使用Depends实现可复用的验证逻辑

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

五、调试与测试技巧

1. 自动文档：访问/docs或/redoc查看交互式文档
2. 请求验证：利用Pydantic模型自动生成验证错误
3. 测试客户端：使用TestClient编写单元测试

from fastapi.testclient import TestClient
client = TestClient(app)
def test_read_item():
    response = client.get("/items/5?q=test")
    assert response.status_code == 200
    assert response.json() == {"item_id": 5, "q": "test"}

六、生产环境实践

1. 中间件配置：添加CORS、GZip等中间件
2. 异常处理：统一异常处理机制
3. 日志记录：集成结构化日志系统

from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": exc.detail},
    )

通过系统掌握上述请求与响应处理技术，开发者可以高效构建出符合RESTful规范的API服务。FastAPI的自动文档、类型验证和异步支持等特性，能显著提升开发效率和代码质量。建议开发者在实际项目中结合这些技术点，逐步构建出健壮的Web服务。