FastAPI接口延迟1800ms降至90ms：Profile与缓存优化实录

import cProfile
import pstats
import io
from functools import wraps

def profile_endpoint(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        pr = cProfile.Profile()
        pr.enable()
        result = func(*args, **kwargs)
        pr.disable()
        s = io.StringIO()
        ps = pstats.Stats(pr, stream=s).sort_stats('cumulative')
        ps.print_stats(20)
        app.logger.info("Profile result:\n%s", s.getvalue())
        return result
    return wrapper

@app.route('/api/v1/customers', methods=['GET'])
@profile_endpoint
def get_customers():
    # ... 原有逻辑

ALTER TABLE orders ADD INDEX idx_customer_id (customer_id);
ALTER TABLE customers ADD INDEX idx_status_updated (status, updated_at DESC);

# main.py - FastAPI版本
from fastapi import FastAPI, Depends, Query
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import selectinload, sessionmaker

DATABASE_URL = "mysql+asyncmy://user:pass@host:3306/crm"
engine = create_async_engine(DATABASE_URL, echo=False, pool_size=20, max_overflow=10)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

app = FastAPI()

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/api/v1/customers")
async def get_customers(
    status: str = Query("active"),
    limit: int = Query(200, le=200),
    db: AsyncSession = Depends(get_db)
):
    stmt = (
        select(Customer)
        .options(selectinload(Customer.orders))
        .where(Customer.status == status)
        .order_by(Customer.updated_at.desc())
        .limit(limit)
    )
    result = await db.execute(stmt)
    customers = result.scalars().all()

    # 批量统计订单金额，避免逐条count
    customer_ids = [c.id for c in customers]
    stats_stmt = (
        select(Order.customer_id, func.sum(Order.amount).label("total"), func.count(Order.id).label("cnt"))
        .where(Order.customer_id.in_(customer_ids))
        .group_by(Order.customer_id)
    )
    stats = await db.execute(stats_stmt)
    stat_map = {row.customer_id: {"total": row.total, "cnt": row.cnt} for row in stats}

    return [{
        "id": c.id,
        "name": c.name,
        "status": c.status,
        "order_stats": stat_map.get(c.id, {"total": 0, "cnt": 0})
    } for c in customers]

import redis.asyncio as aioredis
import json
from fastapi import HTTPException

redis_client = aioredis.from_url("redis://localhost:6379/0", decode_responses=True)

async def get_customer_stats_from_cache(customer_ids):
    """使用Redis pipeline批量获取缓存"""
    pipe = redis_client.pipeline()
    for cid in customer_ids:
        await pipe.get(f"cust_stats:{cid}")
    results = await pipe.execute()

    # 命中率统计
    hit = sum(1 for r in results if r is not None)
    if hit / len(customer_ids) < 0.5:
        print(f"缓存命中率过低: {hit}/{len(customer_ids)}")  # 生产环境用logger

    return results

@app.get("/api/v1/customers")
async def get_customers_cached(
    status: str = Query("active"),
    limit: int = Query(200, le=200),
    db: AsyncSession = Depends(get_db)
):
    # 先查缓存
    cache_key = f"customers_list:{status}:{limit}"
    cached_data = await redis_client.get(cache_key)
    if cached_data:
        return json.loads(cached_data)

    # 数据库查询逻辑同上...
    # 构造response后存入缓存
    await redis_client.setex(cache_key, 60, json.dumps(response_data))
    return response_data