FastAPI接口耗时400ms优化到45ms：Profiling与缓存策略全记录

import cProfile
import pstats
from io import StringIO

# 假设这是你的FastAPI应用
from main import app
from fastapi.testclient import TestClient

client = TestClient(app)

# 模拟一次真实请求，捕获profile数据
profiler = cProfile.Profile()
profiler.enable()
client.get("/api/v1/users/12345/order-stats")
profiler.disable()

# 输出耗时Top 10
s = StringIO()
ps = pstats.Stats(profiler, stream=s).sort_stats("cumulative")
ps.print_stats(10)
print(s.getvalue())

ncalls  tottime  percall  cumtime  percall  filename:lineno(function)
1       0.002    0.002    0.381    0.381  fastapi/routing.py:... (get_route_handler)
83      0.008    0.000    0.247    0.003  sqlalchemy/orm/loading.py:... (_load_scalar_from_orm)
83      0.011    0.000    0.225    0.003  sqlalchemy/dialects/mysql/asyncai.py:... (_execute_context)
1       0.067    0.067    0.067    0.067  json/encoder.py:... (default)

# 安装并获取进程PID
pip install py-spy
pgrep -f "uvicorn main:app"

# 采样10秒，生成火焰图SVG
py-spy record --pid  --duration 10 -o /tmp/perf.svg
py-spy top --pid  --duration 5

# 优化前（懒加载，N+1）
async def get_order_stats(user_id: int):
    orders = await session.execute(
        select(Order).where(Order.user_id == user_id)
    )
    orders = orders.scalars().all()

    total_amount = 0.0
    item_count = 0
    for order in orders:
        for item in order.items:  # 这里每访问一次发一条SQL！
            total_amount += item.price
            item_count += 1
    return {"total": total_amount, "count": item_count}

# 优化后（selectinload，2条SQL）
from sqlalchemy.orm import selectinload

async def get_order_stats(user_id: int):
    orders = await session.execute(
        select(Order)
        .options(selectinload(Order.items))  # 一次性预加载所有items
        .where(Order.user_id == user_id)
    )
    orders = orders.scalars().all()

    total_amount = sum(item.price for order in orders for item in order.items)
    item_count = sum(len(order.items) for order in orders)
    return {"total": total_amount, "count": item_count}

from sqlalchemy import func, select, and_

async def get_order_stats_fast(user_id: int):
    # 单条SQL聚合查询，数据库端完成计算
    result = await session.execute(
        select(
            func.coalesce(func.sum(OrderItem.price), 0.0).label("total"),
            func.count(OrderItem.id).label("count")
        )
        .join(Order, Order.id == OrderItem.order_id)
        .where(and_(Order.user_id == user_id, 
                    Order.created_at >= datetime.now() - timedelta(days=30)))
    )
    row = result.one()
    return {"total": row.total, "count": row.count}

import json
import time
from functools import lru_cache
from fastapi import APIRouter, Depends
from redis.asyncio import Redis

router = APIRouter()

# 本地缓存：简单字典 + 过期时间
_local_cache = {}
_LOCAL_TTL = 30  # 本地缓存30秒

# Redis连接（全局复用）
redis_client = Redis(host="localhost", port=6379, decode_responses=True)
_REDIS_TTL = 60  # Redis缓存60秒

async def get_order_stats_cached(user_id: int):
    # 1. 查本地缓存（最快）
    cached = _local_cache.get(user_id)
    if cached and time.time() - cached[1] < _LOCAL_TTL:
        return cached[0]

    # 2. 查Redis（跨进程共享）
    redis_key = f"order_stats:{user_id}"
    redis_val = await redis_client.get(redis_key)
    if redis_val:
        data = json.loads(redis_val)
        _local_cache[user_id] = (data, time.time())
        return data

    # 3. 查数据库（唯一慢路径）
    data = await query_from_database(user_id)  # 即上一节的聚合SQL

    # 4. 回填缓存
    await redis_client.setex(redis_key, _REDIS_TTL, json.dumps(data))
    _local_cache[user_id] = (data, time.time())
    return data

@router.get("/api/v1/users/{user_id}/order-stats")
async def get_stats(user_id: int):
    return await get_order_stats_cached(user_id)