第一章：动态路由工程化
1.1 正则表达式路由

PYTHON

from fastapi import Path


@app.get("/users/{user_id:int}")
async def get_user(
        user_id: int = Path(..., regex="^[0-9]{8}$", example=10000001)
):
    # 匹配8位数字ID
    return db.query(User).filter(User.id == user_id).first()

1.2 权重路由控制

PYTHON

# 高优先级路由
@app.get("/users/me", priority=100)
async def get_current_user():
    ...


# 低优先级通用路由  
@app.get("/users/{user_id}", priority=10)
async def get_user(user_id: int):
    ...

第二章：复杂请求处理
2.1 多层嵌套请求体

PYTHON

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


class UserProfile(BaseModel):
    name: str
    addresses: list[Address]


@app.post("/users")
async def create_user(profile: UserProfile):
    # 自动解析嵌套结构
    db.save(profile.dict())

2.2 大文件分片上传

PYTHON

from fastapi import UploadFile, File


@app.post("/upload")
async def upload_large_file(
        chunk: UploadFile = File(...),
        chunk_number: int = Form(...)
):
    with open(f"temp_{chunk_number}", "wb") as buffer:
        content = await chunk.read()
        buffer.write(content)
    return {"received_chunks": chunk_number + 1}

第三章：路由性能调优
3.1 依赖注入缓存策略

PYTHON

from fastapi import Depends


def get_db():
    # 数据库连接池
    return DatabasePool()


@app.get("/products")
async def list_products(
        page: int = 1,
        db: Database = Depends(get_db)
):
    return db.query(Product).paginate(page)

3.2 路由惰性加载

PYTHON

# 按需加载路由模块
from fastapi import APIRouter

order_router = APIRouter()


@order_router.get("/")
async def list_orders():
    ...


app.include_router(order_router, prefix="/orders")

第四章：安全加固实战
4.1 路由级速率限制

PYTHON

from fastapi_limiter import Limiter

limiter = Limiter(key_func=get_remote_address)


@app.get("/api/data", dependencies=[Depends(limiter.limit("100/min"))])
async def sensitive_data():
    return generate_report()

4.2 SQL注入终极防御

PYTHON

# 危险：直接拼接
@app.get("/users")
async def unsafe_query(name: str):
    query = f"SELECT * FROM users WHERE name = '{name}'"


# 安全：参数化查询
@app.get("/users")
async def safe_query(name: str = Query(...)):
    query = "SELECT * FROM users WHERE name = :name"
    params = {"name": name}
    return db.execute(query, params)

课后超级实验室
任务1：设计商品SKU路由系统

PYTHON

# 要求：
# 1. 支持SKU编码校验（格式：ABC-12345）
# 2. 实现库存实时扣减
# 3. 处理高并发冲突
@app.put("/skus/{sku_code}")
async def update_sku(sku_code: str, stock: int):
# 你的代码

任务2：优化订单查询性能

PYTHON

# 原代码
@app.get("/orders")
async def list_orders():
    return db.query(Order).all()

# 优化目标：响应时间 <100ms（提示：添加缓存索引）

错误诊疗室

来源：https://blog.cmdragon.cn/posts/8737e29cfe7a/