FastAPI性能调优实录：Profiling定位+SQL优化+Redis缓存，接口从2.1s降到180ms

# profile_api.py
import cProfile
import pstats
import asyncio
from httpx import AsyncClient

async def hit_api():
    async with AsyncClient(base_url="http://127.0.0.1:8000") as client:
        # 模拟真实请求：订单详情 + 用户信息 + 商品列表
        for _ in range(200):  # 跑200次求平均
            r = await client.get("/api/v1/orders/12345?include_items=true")
            assert r.status_code == 200

if __name__ == "__main__":
    profiler = cProfile.Profile()
    profiler.enable()
    asyncio.run(hit_api())
    profiler.disable()
    stats = pstats.Stats(profiler).sort_stats("cumulative")
    stats.print_stats(20)  # 只看前20行

ncalls  tottime  percall  cumtime  percall  filename:lineno(function)
   200    0.002    0.000    4.213    0.021   serializers.py:38 (OrderOutSchema.dump)
   200    0.001    0.000    3.897    0.019   sqlalchemy/orm/loading.py:112 (instance_processor)
  1200    0.045    0.000    2.456    0.002   sqlalchemy/orm/relationships.py:210 (_emit_lazyload)

# repositories/order_repository.py
from sqlalchemy import select
from sqlalchemy.orm import selectinload
from models import Order, OrderItem, User

async def get_order_with_details(session, order_id: int) -> Order:
    stmt = (
        select(Order)
        .options(
            selectinload(Order.user),          # 原来是懒加载，现在强制预加载
            selectinload(Order.items),         # 原来是懒加载，现在强制预加载
        )
        .where(Order.id == order_id)
    )
    result = await session.execute(stmt)
    return result.scalar_one_or_none()

# schemas/order_schema.py
from pydantic import BaseModel

class OrderItemOut(BaseModel):
    sku: str
    price: float
    quantity: int

class OrderOut(BaseModel):
    id: int
    order_no: str
    total_amount: float
    items: list[OrderItemOut]

    # 关键：手动指定字段来源，避免Pydantic遍历ORM对象的所有属性
    @classmethod
    def from_orm_with_items(cls, order: Order) -> "OrderOut":
        return cls(
            id=order.id,
            order_no=order.order_no,
            total_amount=order.total_amount,
            items=[
                OrderItemOut(
                    sku=item.sku,
                    price=item.price,
                    quantity=item.quantity
                )
                for item in order.items
            ]
        )

# utils/cache.py
import json
import hashlib
from redis import asyncio as aioredis

redis_client = aioredis.from_url(
    "redis://127.0.0.1:6379/0",
    decode_responses=True,
    max_connections=50,          # 连接池上限，默认10太小
    socket_connect_timeout=2,    # 连接超时，防止Redis挂了拖垮API
    socket_timeout=2,
)

def cache_response(ttl: int, prefix: str):
    def decorator(func):
        async def wrapper(*args, **kwargs):
            # 生成缓存key：函数名 + 参数hash
            key = f"{prefix}:{hashlib.md5(str(kwargs).encode()).hexdigest()}"
            cached = await redis_client.get(key)
            if cached:
                return json.loads(cached)

            result = await func(*args, **kwargs)
            # 只缓存成功响应，避免缓存空结果
            if result.status_code == 200:
                await redis_client.setex(key, ttl, json.dumps(result.body))
            return result
        return wrapper
    return decorator