FastAPI进阶指南：快速掌握请求与响应的核心机制

FastAPI作为现代Python Web框架的代表，凭借其高性能、类型安全和开发友好的特性，已成为构建RESTful API的首选工具。本文将聚焦于FastAPI中请求与响应的基础用法，通过代码示例和场景分析，帮助开发者快速掌握参数处理、响应封装及中间件应用等核心功能。

一、请求参数处理：从路径到请求体

1. 路径参数与查询参数

路径参数通过花括号{}定义，查询参数通过Query类声明默认值。以下示例展示如何同时处理两种参数：

from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(
    item_id: int,
    q: str = Query(None, title="查询字符串", max_length=50)
):
    return {"item_id": item_id, "q": q}

• 路径参数：item_id通过URL路径传递（如/items/123）
• 查询参数：q通过URL查询字符串传递（如/items/123?q=test）
• 参数验证：Query类可设置默认值、数据类型和长度限制

2. 请求体处理：Pydantic模型的应用

FastAPI通过Pydantic模型自动处理JSON请求体，实现类型安全的参数解析：

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

• 模型定义：Item类定义了请求体的字段结构
• 自动解析：FastAPI自动将JSON请求体转换为Item对象
• 字段扩展：可在视图函数中动态添加计算字段（如price_with_tax）

3. 多参数组合场景

实际开发中常需同时处理路径、查询和请求体参数：

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Item,
    q: str = None
):
    result = {"item_id": item_id, **item.dict()}
    if q:
        result.update({"q": q})
    return result

• 参数优先级：路径参数 > 查询参数 > 请求体参数
• 数据合并：通过字典解包（**item.dict()）实现字段合并

二、响应封装：从原生到结构化

1. 原生响应处理

FastAPI支持直接返回字典、列表等原生Python类型：

@app.get("/items/simple")
async def get_simple_item():
    return {"name": "FastAPI Book", "price": 29.99}

• 自动转换：字典自动转换为JSON响应
• 状态码设置：通过status_code参数指定（如status_code=201）

2. 结构化响应：JSONResponse与ORJSONResponse

对于需要自定义响应头的场景，可使用JSONResponse：

from fastapi.responses import JSONResponse
@app.get("/items/custom")
async def get_custom_item():
    return JSONResponse(
        content={"message": "Custom response"},
        headers={"X-Custom": "FastAPI"}
    )

• 性能优化：ORJSONResponse（需安装orjson）提供更快的JSON序列化
• 二进制响应：StreamingResponse适用于大文件传输

3. 响应模型与字段过滤

通过response_model参数实现响应数据过滤和序列化：

from fastapi import FastAPI
from pydantic import BaseModel
class OutputItem(BaseModel):
    name: str
    price: float
@app.get("/items/model", response_model=OutputItem)
async def get_model_item():
    return {"name": "Filtered Item", "price": 19.99, "description": "Hidden field"}

• 字段过滤：响应中仅包含OutputItem定义的字段
• 数据验证：自动验证返回数据是否符合模型定义

三、中间件：请求响应的全局处理

1. 中间件基础实现

中间件可在请求处理前后执行自定义逻辑：

from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def log_requests(request: Request, call_next):
    print(f"Request path: {request.url.path}")
    response = await call_next(request)
    print(f"Response status: {response.status_code}")
    return response

• 执行顺序：请求到达时先执行中间件，再调用路由处理函数
• 响应修改：可在返回前修改响应头或状态码

2. 性能监控中间件示例

以下中间件实现请求耗时统计：

import time
from fastapi import Request
@app.middleware("http")
async def add_timing_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

• 性能指标：通过X-Process-Time响应头暴露处理耗时
• 无侵入设计：不修改路由处理逻辑即可添加功能

四、高级场景实践

1. 异步请求处理

FastAPI原生支持异步视图函数，适用于IO密集型操作：

import httpx
@app.get("/external-api")
async def call_external_api():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
    return response.json()

• 并发优势：异步请求可避免阻塞事件循环
• 超时控制：建议配合timeout参数使用（如httpx.AsyncClient(timeout=5.0)）

2. 请求验证与错误处理

通过HTTPException实现标准化错误响应：

from fastapi import HTTPException
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id < 0:
        raise HTTPException(
            status_code=400,
            detail="Item ID must be positive"
        )
    return {"item_id": item_id}

• 错误码规范：4xx表示客户端错误，5xx表示服务器错误
• 自定义异常：可继承HTTPException实现业务特定错误

3. 依赖注入系统

FastAPI的依赖注入系统可简化参数共享：

from fastapi import Depends
def query_validator(q: str = Query(...)):
    if len(q) < 3:
        raise HTTPException(status_code=400, detail="Query too short")
    return q
@app.get("/items/dependency")
async def read_items(q: str = Depends(query_validator)):
    return [{"item": f"Item {q}"}]

• 复用逻辑：将验证逻辑封装为依赖项
• 作用域控制：支持request、session等不同生命周期的依赖

五、最佳实践建议

1. 参数验证优先：始终为路径参数和查询参数设置类型和约束
2. 响应模型分离：区分请求模型（Item）和响应模型（OutputItem）
3. 中间件分层：将日志、监控等横切关注点封装为中间件
4. 异步优先：对于外部API调用等IO操作，优先使用异步方式
5. 文档增强：利用FastAPI自动生成的OpenAPI文档，通过description和example参数丰富文档内容

通过掌握上述请求与响应的核心机制，开发者可以高效构建类型安全、性能优异的RESTful API。FastAPI的自动文档、数据验证和异步支持等特性，将显著提升开发效率和代码质量。