Python 异步 Web 框架：FastAPI 实战

FastAPI 是一个现代、高性能的 Python Web 框架，内置异步支持，基于 Starlette 核心和 Pydantic 数据验证。与传统同步框架 Django/Flask 不同，FastAPI 的路由函数默认异步执行，能够充分发挥 Python asyncio 的并发优势，同时自动生成 OpenAPI 文档，极大提升 API 开发效率。

FastAPI 的核心优势：高性能（与 Node.js 和 Go 相当）、自动数据验证（Pydantic）、自动文档生成、异步原生支持、依赖注入系统。

快速入门

from fastapi import FastAPI
from pydantic import BaseModel
import asyncio
app = FastAPI()
class Item(BaseModel):
    name: str
    price: float
    tags: list[str] = []
GET 请求
@app.get("/")
async def root():
    return {"message": "Hello FastAPI"}
路径参数
@app.get("/items/{item_id}")
async def get_item(item_id: int):
    return {"item_id": item_id, "name": "Sample Item"}
请求体
@app.post("/items/")
async def create_item(item: Item):
    # Pydantic 自动验证数据
    return {"created": item.dict()}
查询参数
@app.get("/search")
async def search(q: str = "", limit: int = 10):
    return {"query": q, "limit": limit}
启动服务
uvicorn main:app --reload --host 0.0.0.0 --port 8000</code></pre>
异步数据库集成
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import Column, Integer, String
from pydantic import BaseModel
Base = declarative_base()
engine = create_async_engine("sqlite+aiosqlite:///./test.db")
class ItemModel(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
依赖注入：获取数据库会话
async def get_db():
    async with async_session() as session:
        yield session
class ItemCreate(BaseModel):
    name: str
class ItemResponse(BaseModel):
    id: int
    name: str
@app.post("/items/", response_model=ItemResponse)
async def create_item(item: ItemCreate, db: AsyncSession = Depends(get_db)):
    db_item = ItemModel(name=item.name)
    db.add(db_item)
    await db.commit()
    await db.refresh(db_item)
    return db_item
@app.get("/items/", response_model=list[ItemResponse])
async def list_items(db: AsyncSession = Depends(get_db)):
    result = await db.execute("SELECT * FROM items")
    items = result.fetchall()
    return [ItemResponse(id=i[0], name=i[1]) for i in items]

并发请求处理
import httpx
from fastapi import FastAPI
app = FastAPI()
@app.get("/aggregate")
async def aggregate_data():
    """并发调用多个外部服务"""
    async with httpx.AsyncClient() as client:
        # 同时请求多个接口
        users, orders, products = await asyncio.gather(
            client.get("https://api.example.com/users"),
            client.get("https://api.example.com/orders"),
            client.get("https://api.example.com/products"),
        )
        return {
            "users": users.json(),
            "orders": orders.json(),
            "products": products.json(),
        }

后台任务
from fastapi import BackgroundTasks
def send_notification(email: str, message: str):
    """同步函数，在后台线程池执行"""
    import time
    time.sleep(2)
    print(f"通知已发送至 {email}: {message}")
@app.post("/orders/")
async def create_order(background_tasks: BackgroundTasks, email: str):
    order = {"order_id": 12345, "status": "created"}
    # 注册后台任务，不阻塞响应
    background_tasks.add_task(send_notification, email, "订单已创建")
    return order

WebSocket 支持
from fastapi import WebSocket
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"收到: {data}")
    except Exception:
        await websocket.close()

中间件与异常处理
from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    import time
    start = time.time()
    response = await call_next(request)
    response.headers["X-Process-Time"] = str(time.time() - start)
    return response
@app.exception_handler(HTTPException)
async def custom_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": exc.detail},
    )

注意事项

FastAPI 路由函数应使用 async def，但如果内部调用同步库（如同步数据库驱动），会导致事件循环阻塞，此时应使用普通 def，FastAPI 会自动在线程池中运行
Pydantic v2 使用 BaseModel 的 model_validate 和 model_dump 方法，而非 .dict()
生产环境推荐使用 Gunicorn + Uvicorn Worker：gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker
使用  lifespan  上下文管理器管理启动和关闭事件

FastAPI 以其高性能、自动文档和现代 Python 特性，已成为 Python 异步 Web 开发的首选框架，特别适合构建微服务、RESTful API 和实时通信服务。