第 4 章：FastAPI 框架详解¶

┌─────────────────────────────────────────┐
│              FastAPI                     │
│                                          │
│  ┌─────────────┐  ┌──────────────────┐  │
│  │  Starlette  │  │   Pydantic       │  │
│  │  (Web 层)   │  │  (数据验证层)     │  │
│  │  异步路由    │  │  类型验证        │  │
│  │  WebSocket  │  │  序列化          │  │
│  └─────────────┘  └──────────────────┘  │
│                                          │
│  ┌──────────────────────────────────┐   │
│  │  OpenAPI (Swagger) 自动生成       │   │
│  └──────────────────────────────────┘   │
└─────────────────────────────────────────┘

pip install fastapi uvicorn

from fastapi import FastAPI

app = FastAPI(
    title="我的 API",
    description="这是一个示例 API",
    version="1.0.0"
)

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

@app.get("/users/{user_id}")
def get_user(user_id: int, name: str = "Anonymous"):
    return {"user_id": user_id, "name": name}

uvicorn main:app --reload

# main:app → 文件名:应用实例名
# --reload → 开发模式，代码修改自动重启

from fastapi import FastAPI, Path
from typing import Annotated

app = FastAPI()

# 基本路径参数
@app.get("/users/{user_id}")
def get_user(user_id: int):
    # user_id 自动转换为 int，如果不是整数会返回 422
    return {"user_id": user_id, "type": type(user_id).__name__}

# 带验证的路径参数
@app.get("/items/{item_id}")
def get_item(
    item_id: Annotated[int, Path(title="物品 ID", ge=1, le=1000)]
):
    return {"item_id": item_id}

# 多个路径参数
@app.get("/users/{user_id}/posts/{post_id}")
def get_user_post(user_id: int, post_id: int):
    return {"user_id": user_id, "post_id": post_id}

# 查询参数自动识别（不是路径参数的都是查询参数）
@app.get("/items")
def list_items(
    skip: int = 0,        # 默认值
    limit: int = 100,
    q: str | None = None  # 可选参数
):
    results = [{"id": i} for i in range(skip, skip + limit)]
    if q:
        results = [r for r in results if q in str(r)]
    return {"items": results, "query": q}

# 访问：GET /items?skip=0&limit=10&q=test

# 没有默认值的参数 = 必需参数
@app.get("/search")
def search(q: str, page: int = 1):
    # q 是必需的，page 是可选的（有默认值）
    return {"q": q, "page": page}

# 访问：GET /search?q=python  ✅
# 访问：GET /search           ❌ 422 Unprocessable Entity

from pydantic import BaseModel, EmailStr, Field
from typing import Optional

class UserCreate(BaseModel):
    name: str
    email: str
    age: int
    bio: Optional[str] = None  # 可选字段

@app.post("/users")
def create_user(user: UserCreate):
    # FastAPI 自动：
    # 1. 解析 JSON 请求体
    # 2. 验证类型（name 必须是字符串，age 必须是整数）
    # 3. 返回 Pydantic 模型实例
    return {"message": "用户创建成功", "user": user.model_dump()}

from pydantic import BaseModel, Field, field_validator
from datetime import date

class UserUpdate(BaseModel):
    name: str = Field(min_length=1, max_length=50, description="用户名")
    email: str = Field(pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    age: int = Field(ge=0, le=150, description="年龄")
    birth_date: date | None = None

    @field_validator('name')
    @classmethod def name_must_not_be_empty(cls, v):
        if not v.strip():
            raise ValueError('用户名不能为空')
        return v.strip()

    @field_validator('age')
    @classmethod def age_must_be_positive(cls, v):
        if v < 0:
            raise ValueError('年龄不能为负数')
        return v

class Address(BaseModel):
    street: str
    city: str
    country: str
    zipcode: str

class UserDetail(BaseModel):
    name: str
    email: str
    address: Address  # 嵌套模型
    tags: list[str] = []

@app.post("/users/detailed")
def create_detailed_user(user: UserDetail):
    return user.model_dump()

# 请求体示例：
# {
#   "name": "Alice",
#   "email": "alice@example.com",
#   "address": {
#     "street": "123 Main St",
#     "city": "Beijing",
#     "country": "China",
#     "zipcode": "100000"
#   },
#   "tags": ["developer", "python"]
# }

from pydantic import BaseModel, Field

class UserInDB(BaseModel):
    id: int
    name: str
    email: str
    hashed_password: str  # 数据库中有密码
    created_at: str

class UserPublic(BaseModel):
    """对外暴露的模型（不包含密码）"""
    id: int
    name: str
    email: str
    # 没有 hashed_password！

@app.post("/users", response_model=UserPublic)
def create_user(user: UserCreate):
    # 即使返回完整数据，响应也会只包含 UserPublic 的字段
    db_user = save_to_db(user)  # 包含 hashed_password
    return db_user  # 自动过滤掉 hashed_password

from fastapi.responses import JSONResponse

@app.get("/users/{user_id}",
    response_model=UserPublic,
    responses={
        404: {"model": {"detail": str}},
        422: {"description": "验证错误"}
    }
)
def get_user(user_id: int):
    user = find_user(user_id)
    if not user:
        return JSONResponse(status_code=404, content={"detail": "用户不存在"})
    return user

# 同步路由（在线程池中运行，不阻塞 Event Loop）
@app.get("/sync")
def sync_endpoint():
    # 适合 CPU 密集型操作
    result = heavy_computation()
    return {"result": result}

# 异步路由（直接在 Event Loop 中运行）
@app.get("/async")
async def async_endpoint():
    # 适合 I/O 密集型操作
    result = await fetch_from_database()
    result2 = await call_external_api()
    return {"result": result, "result2": result2}

import httpx

@app.get("/aggregate")
async def aggregate_data():
    async with httpx.AsyncClient() as client:
        # 并发请求，不是串行！
        user_resp, post_resp, comment_resp = await asyncio.gather(
            client.get("http://users-api/users/1"),
            client.get("http://posts-api/posts?user_id=1"),
            client.get("http://comments-api/comments?user_id=1"),
        )
    return {
        "user": user_resp.json(),
        "posts": post_resp.json(),
        "comments": comment_resp.json(),
    }

from fastapi import Depends, Query

# 定义依赖
async def common_params(
    skip: int = Query(0, ge=0),
    limit: int = Query(100, ge=1, le=100),
):
    return {"skip": skip, "limit": limit}

# 使用依赖
@app.get("/items")
async def list_items(params: dict = Depends(common_params)):
    return {"params": params}

@app.get("/users")
async def list_users(params: dict = Depends(common_params)):
    return {"params": params}

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security)
):
    token = credentials.credentials
    user = decode_token(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效的认证令牌"
        )
    return user

@app.get("/users/me")
async def get_me(current_user: dict = Depends(get_current_user)):
    return current_user

from sqlalchemy.orm import Session

def get_db():
    db = SessionLocal()
    try:
        yield db  # 生成器依赖：yield 之前的代码在请求前执行，yield 之后的代码在请求后执行
    finally:
        db.close()

@app.get("/items/{item_id}")
def get_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(Item).filter(Item.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="物品不存在")
    return item

async def verify_token(token: str = Depends(security)):
    return decode_token(token.credentials)

async def get_user_from_token(
    token_data: dict = Depends(verify_token)
):
    user = find_user(token_data["user_id"])
    if not user:
        raise HTTPException(status_code=401, detail="用户不存在")
    return user

async def require_admin(
    current_user: dict = Depends(get_user_from_token)
):
    if not current_user.get("is_admin"):
        raise HTTPException(status_code=403, detail="需要管理员权限")
    return current_user

@app.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    admin: dict = Depends(require_admin)
):
    # 只有管理员能访问
    ...

from starlette.middleware.base import BaseHTTPMiddleware
import time

# 请求耗时中间件
@app.middleware("http")
async def process_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

# CORS 中间件
from starlette.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

from fastapi import WebSocket

@app.websocket("/ws/chat")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            # 广播给所有连接
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        pass

my-fastapi-app/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI 应用入口
│   ├── config.py            # 配置
│   ├── models/
│   │   ├── __init__.py
│   │   ├── user.py          # Pydantic 模型
│   │   └── post.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── user.py          # 请求/响应 Schema
│   │   └── post.py
│   ├── api/
│   │   ├── __init__.py
│   │   ├── deps.py          # 依赖注入
│   │   ├── users.py         # 用户路由
│   │   └── posts.py         # 文章路由
│   ├── services/
│   │   ├── user_service.py  # 业务逻辑
│   │   └── email_service.py
│   ├── database.py          # 数据库连接
│   └── middleware.py        # 中间件
├── tests/
│   ├── test_users.py
│   └── test_posts.py
├── requirements.txt
├── Dockerfile
└── README.md

from fastapi import FastAPI, File, UploadFile
from pydantic import BaseModel
import numpy as np

app = FastAPI(title="AI 图像分类服务")

# 加载模型（启动时加载一次）
model = load_model("resnet50.pth")
classes = ["cat", "dog", "bird", "fish"]

class PredictionResult(BaseModel):
    class_name: str
    confidence: float
    all_probs: dict

@app.post("/predict", response_model=PredictionResult)
async def predict(file: UploadFile = File(...)):
    # 读取上传的图片
    contents = await file.read()
    image = preprocess_image(contents)

    # 模型推理
    with torch.no_grad():
        output = model(image)
        probs = torch.softmax(output, dim=1)[0]

    # 返回结果
    top_idx = probs.argmax().item()
    return PredictionResult(
        class_name=classes[top_idx],
        confidence=probs[top_idx].item(),
        all_probs={classes[i]: probs[i].item() for i in range(len(classes))}
    )

@app.get("/health")
async def health():
    return {"status": "healthy", "model_loaded": model is not None}