FastAPI 构建高效 API 服务的完整方法指南，涵盖从基础到进阶的实践技巧：

---

1. 核心特性

• 高性能：基于 Starlette（异步）和 Pydantic（数据校验）
• 开发效率：自动生成 API 文档（Swagger/ReDoc）
• 生产就绪：支持 WebSocket、后台任务、依赖注入

---

2. 基础使用

安装

pip install fastapi uvicorn[standard]

最小示例

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Hello World"}

启动服务

uvicorn main:app --reload  # --reload 启用热更新（开发模式）

---

3. 核心功能

路由与请求参数

from fastapi import Path, Query


@app.get("/items/{item_id}")
async def read_item(
        item_id: int = Path(..., title="商品ID", gt=0),  # 路径参数
        q: str = Query(None, min_length=3, max_length=50)  # 查询参数
):
    return {"item_id": item_id, "q": q}

请求体与数据校验

from pydantic import BaseModel


class Item(BaseModel):
    name: str
    price: float
    tags: list[str] = []


@app.post("/items/")
async def create_item(item: Item):
    return item

异步支持

import httpx


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

---

4. 进阶技巧

依赖注入

from fastapi import Depends


def query_extractor(q: str = Query(...)):
    return q


@app.get("/dependency/")
async def use_dependency(data: str = Depends(query_extractor)):
    return {"data": data}

中间件

@app.middleware("http")
async def add_process_time_header(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

后台任务

from fastapi import BackgroundTasks


def write_log(message: str):
    with open("log.txt", "a") as f:
        f.write(message)


@app.post("/send-email")
async def send_email(
        email: str,
        background_tasks: BackgroundTasks
):
    background_tasks.add_task(write_log, f"Email sent to {email}")
    return {"message": "Email queued"}

---

5. 部署优化

生产环境启动

# 使用 Gunicorn + Uvicorn Worker
gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app

Docker 部署

FROM python:3.9-slim

RUN pip install fastapi uvicorn gunicorn

COPY ./app /app

CMD ["gunicorn", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "app.main:app"]

性能优化

• 静态文件缓存：

  from fastapi.staticfiles import StaticFiles
  
  app.mount("/static", StaticFiles(directory="static"), name="static")

• 请求超时控制：在反向代理（Nginx）中设置

  proxy_read_timeout 300s;

---

6. 结合机器学习模型

模型服务化示例

from fastapi import File, UploadFile
import numpy as np
from PIL import Image
import io

# 伪代码示例：加载预训练模型
model = load_model("model.pkl")


@app.post("/predict")
async def predict(image: UploadFile = File(...)):
    # 处理上传的图片
    contents = await image.read()
    img = Image.open(io.BytesIO(contents)).convert("RGB")
    img_array = preprocess(img)

    # 模型预测
    prediction = model.predict(img_array[np.newaxis, ...])
    return {"class": prediction[0]}

最佳实践

1. 模型预热：在启动时加载模型到内存
2. 批处理支持：使用 asyncio 队列实现请求批处理
3. 监控集成：通过 Prometheus 暴露指标

   from starlette_exporter import PrometheusMiddleware
   
   app.add_middleware(PrometheusMiddleware)

---

7. 文档与测试

自动生成的 API 文档

• Swagger UI：http://localhost:8000/docs
• ReDoc：http://localhost:8000/redoc

单元测试

from fastapi.testclient import TestClient

client = TestClient(app)


def test_read_item():
    response = client.get("/items/42?q=test")
    assert response.status_code == 200
    assert response.json() == {"item_id": 42, "q": "test"}

---

常见问题解决

1. CORS 问题：添加中间件

   from fastapi.middleware.cors import CORSMiddleware
   app.add_middleware(
       CORSMiddleware,
       allow_origins=["*"],
       allow_methods=["*"],
   )

2. 大文件上传：设置请求体大小限制

   app = FastAPI(max_request_size=1024 * 1024 * 100)  # 100MB

---

通过以上方法，你可以快速构建高性能 API 服务，并轻松集成 AI 模型！建议结合 官方文档 深入学习高级功能。