1. FastAPI如何实现日志记录？

from fastapi import FastAPI, Request
import logging
import json
from datetime import datetime

app = FastAPI()

# 日志配置
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# 结构化日志
import structlog

structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer()
    ]
)

structured_logger = structlog.get_logger()

# 请求日志中间件
@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()
    
    # 请求日志
    logger.info(f"Request: {request.method} {request.url}")
    
    response = await call_next(request)
    
    # 响应日志
    process_time = time.time() - start_time
    logger.info(f"Response: {response.status_code} - {process_time:.3f}s")
    
    return response

# 结构化日志中间件
@app.middleware("http")
async def structured_logging(request: Request, call_next):
    start_time = time.time()
    
    structured_logger.info(
        "request_started",
        method=request.method,
        path=request.url.path,
        client=request.client.host
    )
    
    response = await call_next(request)
    
    structured_logger.info(
        "request_completed",
        status_code=response.status_code,
        duration=time.time() - start_time
    )
    
    return response

# 异常日志
@app.exception_handler(Exception)
async def log_exception(request: Request, exc: Exception):
    logger.exception(f"Unhandled exception: {exc}")
    return JSONResponse(
        status_code=500,
        content={"error": "Internal server error"}
    )

# 审计日志
def audit_log(action: str, user_id: int, details: dict):
    log_entry = {
        "timestamp": datetime.now().isoformat(),
        "action": action,
        "user_id": user_id,
        "details": details
    }
    logger.info(f"AUDIT: {json.dumps(log_entry)}")

@app.post("/users")
async def create_user(user: UserCreate):
    result = save_user(user)
    audit_log("create_user", user.id, {"email": user.email})
    return result

2. FastAPI如何实现性能监控？

from fastapi import FastAPI, Request
from prometheus_client import Counter, Histogram, generate_latest
from fastapi.responses import PlainTextResponse
import time

app = FastAPI()

# Prometheus指标
REQUEST_COUNT = Counter(
    'http_requests_total',
    'Total HTTP requests',
    ['method', 'endpoint', 'status']
)

REQUEST_LATENCY = Histogram(
    'http_request_duration_seconds',
    'HTTP request latency',
    ['method', 'endpoint']
)

ACTIVE_REQUESTS = Counter(
    'http_requests_active',
    'Active HTTP requests',
    ['method', 'endpoint']
)

# 监控中间件
@app.middleware("http")
async def prometheus_middleware(request: Request, call_next):
    method = request.method
    endpoint = request.url.path
    
    ACTIVE_REQUESTS.labels(method=method, endpoint=endpoint).inc()
    
    start_time = time.time()
    
    response = await call_next(request)
    
    duration = time.time() - start_time
    
    REQUEST_COUNT.labels(
        method=method,
        endpoint=endpoint,
        status=response.status_code
    ).inc()
    
    REQUEST_LATENCY.labels(
        method=method,
        endpoint=endpoint
    ).observe(duration)
    
    ACTIVE_REQUESTS.labels(method=method, endpoint=endpoint).dec()
    
    return response

# Prometheus端点
@app.get("/metrics")
async def metrics():
    return PlainTextResponse(generate_latest())

# 自定义指标
DB_QUERY_COUNT = Counter(
    'db_queries_total',
    'Total database queries',
    ['operation']
)

CACHE_HIT_COUNT = Counter(
    'cache_hits_total',
    'Total cache hits'
)

CACHE_MISS_COUNT = Counter(
    'cache_misses_total',
    'Total cache misses'
)

# 性能追踪
import functools
import time

def timing_decorator(func):
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        start_time = time.time()
        result = await func(*args, **kwargs)
        duration = time.time() - start_time
        
        logger.info(f"{func.__name__} took {duration:.3f}s")
        
        return result
    return wrapper

@app.get("/slow-endpoint")
@timing_decorator
async def slow_endpoint():
    await asyncio.sleep(1)
    return {"message": "Done"}

# 健康检查
@app.get("/health")
async def health_check():
    return {
        "status": "healthy",
        "timestamp": time.time(),
        "version": "1.0.0"
    }

3. FastAPI如何实现分布式追踪？

from fastapi import FastAPI, Request
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger import JaegerExporter
from opentelemetry.sdk.resources import Resource

app = FastAPI()

# 配置追踪
resource = Resource.create({"service.name": "fastapi-app"})
trace.set_tracer_provider(TracerProvider(resource=resource))

jaeger_exporter = JaegerExporter(
    agent_host_name="localhost",
    agent_port=6831,
)

trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(jaeger_exporter)
)

tracer = trace.get_tracer(__name__)

# 追踪中间件
@app.middleware("http")
async def tracing_middleware(request: Request, call_next):
    with tracer.start_as_current_span(
        f"{request.method} {request.url.path}"
    ) as span:
        span.set_attribute("http.method", request.method)
        span.set_attribute("http.url", str(request.url))
        span.set_attribute("http.host", request.headers.get("host", ""))
        
        response = await call_next(request)
        
        span.set_attribute("http.status_code", response.status_code)
        
        return response

# 追踪函数
@app.get("/traced-operation")
async def traced_operation():
    with tracer.start_as_current_span("database_query"):
        result = await query_database()
    
    with tracer.start_as_current_span("external_api_call"):
        data = await call_external_api()
    
    return {"result": result, "data": data}

# 请求ID追踪
from contextvars import ContextVar
import uuid

request_id_var: ContextVar[str] = ContextVar('request_id', default='')

@app.middleware("http")
async def request_id_middleware(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    request_id_var.set(request_id)
    
    response = await call_next(request)
    
    response.headers["X-Request-ID"] = request_id
    
    return response

def get_request_id() -> str:
    return request_id_var.get()

# 日志关联
class RequestIDFilter(logging.Filter):
    def filter(self, record):
        record.request_id = get_request_id()
        return True

logging.basicConfig(
    format='%(asctime)s - [%(request_id)s] - %(levelname)s - %(message)s'
)
logging.getLogger().addFilter(RequestIDFilter())

4. FastAPI如何实现告警系统？

from fastapi import FastAPI
import asyncio
from typing import List
from enum import Enum

app = FastAPI()

class AlertLevel(str, Enum):
    INFO = "info"
    WARNING = "warning"
    ERROR = "error"
    CRITICAL = "critical"

class Alert:
    def __init__(
        self,
        level: AlertLevel,
        message: str,
        details: dict = None
    ):
        self.level = level
        self.message = message
        self.details = details or {}
        self.timestamp = datetime.now()

# 告警处理器
class AlertHandler:
    def __init__(self):
        self.handlers = []
    
    def add_handler(self, handler):
        self.handlers.append(handler)
    
    async def send(self, alert: Alert):
        for handler in self.handlers:
            try:
                await handler(alert)
            except Exception as e:
                logger.error(f"Alert handler failed: {e}")

alert_handler = AlertHandler()

# 邮件告警
async def email_alert(alert: Alert):
    if alert.level in [AlertLevel.ERROR, AlertLevel.CRITICAL]:
        send_email(
            to="admin@example.com",
            subject=f"[{alert.level.value.upper()}] {alert.message}",
            body=json.dumps(alert.details, indent=2)
        )

# Slack告警
async def slack_alert(alert: Alert):
    import httpx
    
    webhook_url = "https://hooks.slack.com/services/xxx"
    
    async with httpx.AsyncClient() as client:
        await client.post(webhook_url, json={
            "text": f"[{alert.level.value.upper()}] {alert.message}",
            "attachments": [{
                "text": json.dumps(alert.details, indent=2)
            }]
        })

alert_handler.add_handler(email_alert)
alert_handler.add_handler(slack_alert)

# 错误监控
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    alert = Alert(
        level=AlertLevel.ERROR,
        message=str(exc),
        details={
            "path": request.url.path,
            "method": request.method,
            "exception_type": type(exc).__name__
        }
    )
    
    await alert_handler.send(alert)
    
    return JSONResponse(
        status_code=500,
        content={"error": "Internal server error"}
    )

# 性能告警
@app.middleware("http")
async def performance_alert(request: Request, call_next):
    start_time = time.time()
    
    response = await call_next(request)
    
    duration = time.time() - start_time
    
    if duration > 5:  # 超过5秒
        alert = Alert(
            level=AlertLevel.WARNING,
            message=f"Slow request: {request.url.path}",
            details={
                "duration": duration,
                "method": request.method
            }
        )
        await alert_handler.send(alert)
    
    return response

# 健康检查告警
async def health_check_monitor():
    while True:
        try:
            async with httpx.AsyncClient() as client:
                response = await client.get("http://localhost:8000/health")
                
                if response.status_code != 200:
                    alert = Alert(
                        level=AlertLevel.CRITICAL,
                        message="Health check failed",
                        details={"status_code": response.status_code}
                    )
                    await alert_handler.send(alert)
        except Exception as e:
            alert = Alert(
                level=AlertLevel.CRITICAL,
                message="Health check exception",
                details={"error": str(e)}
            )
            await alert_handler.send(alert)
        
        await asyncio.sleep(60)

@app.on_event("startup")
async def start_health_monitor():
    asyncio.create_task(health_check_monitor())

5. FastAPI如何实现微服务架构？

from fastapi import FastAPI
import httpx
import asyncio

# 服务网关
app = FastAPI()

# 服务注册
SERVICES = {
    "user-service": "http://localhost:8001",
    "order-service": "http://localhost:8002",
    "product-service": "http://localhost:8003"
}

# 服务路由
@app.api_route("/{service_name}/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def proxy_request(service_name: str, path: str, request: Request):
    if service_name not in SERVICES:
        raise HTTPException(status_code=404, detail="Service not found")
    
    service_url = SERVICES[service_name]
    
    async with httpx.AsyncClient() as client:
        response = await client.request(
            method=request.method,
            url=f"{service_url}/{path}",
            headers=dict(request.headers),
            content=await request.body()
        )
    
    return Response(
        content=response.content,
        status_code=response.status_code,
        headers=dict(response.headers)
    )

# 服务聚合
@app.get("/user-orders/{user_id}")
async def get_user_orders(user_id: int):
    async with httpx.AsyncClient() as client:
        # 并发调用多个服务
        user_task = client.get(f"{SERVICES['user-service']}/users/{user_id}")
        orders_task = client.get(f"{SERVICES['order-service']}/orders?user_id={user_id}")
        
        user_response, orders_response = await asyncio.gather(user_task, orders_task)
    
    return {
        "user": user_response.json(),
        "orders": orders_response.json()
    }

# 服务熔断
from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=30)
async def call_user_service(user_id: int):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"{SERVICES['user-service']}/users/{user_id}")
        return response.json()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    try:
        user = await call_user_service(user_id)
        return user
    except Exception:
        return {"error": "Service unavailable"}

6. FastAPI如何实现服务间通信？

from fastapi import FastAPI
import httpx
import asyncio
from typing import Optional

app = FastAPI()

# HTTP客户端
class ServiceClient:
    def __init__(self, base_url: str, timeout: int = 30):
        self.base_url = base_url
        self.client = httpx.AsyncClient(timeout=timeout)
    
    async def get(self, path: str, params: dict = None):
        response = await self.client.get(f"{self.base_url}{path}", params=params)
        return response.json()
    
    async def post(self, path: str, data: dict):
        response = await self.client.post(f"{self.base_url}{path}", json=data)
        return response.json()
    
    async def close(self):
        await self.client.aclose()

user_client = ServiceClient("http://localhost:8001")
order_client = ServiceClient("http://localhost:8002")

# 同步调用
@app.get("/sync-call")
async def sync_call(user_id: int):
    user = await user_client.get(f"/users/{user_id}")
    orders = await order_client.get(f"/orders", {"user_id": user_id})
    return {"user": user, "orders": orders}

# 并发调用
@app.get("/concurrent-call")
async def concurrent_call(user_id: int):
    user_task = user_client.get(f"/users/{user_id}")
    orders_task = order_client.get(f"/orders", {"user_id": user_id})
    
    user, orders = await asyncio.gather(user_task, orders_task)
    return {"user": user, "orders": orders}

# 消息队列通信
import aio_pika

async def send_message(queue: str, message: dict):
    connection = await aio_pika.connect_robust("amqp://localhost")
    async with connection:
        channel = await connection.channel()
        await channel.declare_queue(queue, durable=True)
        
        await channel.default_exchange.publish(
            aio_pika.Message(body=json.dumps(message).encode()),
            routing_key=queue
        )

@app.post("/async-message")
async def send_async_message(data: dict):
    await send_message("task_queue", data)
    return {"message": "Message sent"}

# gRPC通信
import grpc
from generated import user_pb2, user_pb2_grpc

async def get_user_grpc(user_id: int):
    async with grpc.aio.insecure_channel('localhost:50051') as channel:
        stub = user_pb2_grpc.UserServiceStub(channel)
        response = await stub.GetUser(user_pb2.GetUserRequest(id=user_id))
        return {"id": response.id, "name": response.name}

7. FastAPI如何实现服务熔断？

from fastapi import FastAPI
from circuitbreaker import circuit
import httpx
from enum import Enum

app = FastAPI()

# 使用circuitbreaker库
@circuit(failure_threshold=5, recovery_timeout=30)
async def call_external_service(url: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        response.raise_for_status()
        return response.json()

# 自定义熔断器
class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: int = 30,
        success_threshold: int = 2
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold
        self.failure_count = 0
        self.success_count = 0
        self.state = CircuitState.CLOSED
        self.last_failure_time = None
    
    async def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if self._should_attempt_reset():
                self.state = CircuitState.HALF_OPEN
            else:
                raise Exception("Circuit breaker is open")
        
        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise e
    
    def _should_attempt_reset(self) -> bool:
        return (
            self.last_failure_time and
            time.time() - self.last_failure_time >= self.recovery_timeout
        )
    
    def _on_success(self):
        self.failure_count = 0
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.success_threshold:
                self.state = CircuitState.CLOSED
                self.success_count = 0
    
    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.OPEN
            self.success_count = 0
        elif self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

# 使用自定义熔断器
breaker = CircuitBreaker()

@app.get("/protected-call")
async def protected_call():
    try:
        result = await breaker.call(call_external_api)
        return result
    except Exception as e:
        return {"error": str(e)}

# 熔断器状态监控
@app.get("/circuit-status")
async def circuit_status():
    return {
        "state": breaker.state.value,
        "failure_count": breaker.failure_count,
        "success_count": breaker.success_count
    }

8. FastAPI如何实现服务降级？

from fastapi import FastAPI
import httpx
from functools import wraps

app = FastAPI()

# 降级装饰器
def fallback(fallback_func):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            try:
                return await func(*args, **kwargs)
            except Exception as e:
                return await fallback_func(*args, **kwargs)
        return wrapper
    return decorator

# 默认降级响应
async def default_fallback():
    return {"message": "Service temporarily unavailable", "cached": True}

# 使用降级
@app.get("/users/{user_id}")
@fallback(default_fallback)
async def get_user(user_id: int):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"http://user-service/users/{user_id}")
        response.raise_for_status()
        return response.json()

# 缓存降级
import redis.asyncio as redis

redis_client = redis.from_url("redis://localhost")

async def cached_fallback(key: str):
    cached = await redis_client.get(key)
    if cached:
        return {"data": json.loads(cached), "cached": True}
    return {"error": "No cached data available"}

@app.get("/products/{product_id}")
async def get_product(product_id: int):
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(f"http://product-service/products/{product_id}")
            data = response.json()
            
            # 更新缓存
            await redis_client.setex(
                f"product:{product_id}",
                300,
                json.dumps(data)
            )
            
            return data
    except Exception:
        return await cached_fallback(f"product:{product_id}")

# 功能降级
class FeatureFlags:
    def __init__(self):
        self.flags = {
            "recommendations": True,
            "reviews": True,
            "related_products": True
        }
    
    def is_enabled(self, feature: str) -> bool:
        return self.flags.get(feature, False)
    
    def disable(self, feature: str):
        self.flags[feature] = False
    
    def enable(self, feature: str):
        self.flags[feature] = True

feature_flags = FeatureFlags()

@app.get("/product-details/{product_id}")
async def get_product_details(product_id: int):
    result = {"product": await get_product(product_id)}
    
    if feature_flags.is_enabled("recommendations"):
        try:
            result["recommendations"] = await get_recommendations(product_id)
        except Exception:
            feature_flags.disable("recommendations")
    
    if feature_flags.is_enabled("reviews"):
        try:
            result["reviews"] = await get_reviews(product_id)
        except Exception:
            feature_flags.disable("reviews")
    
    return result

9. FastAPI如何处理高并发请求？

from fastapi import FastAPI
import asyncio
from concurrent.futures import ThreadPoolExecutor
import uvicorn

app = FastAPI()

# 使用uvicorn多worker
# uvicorn main:app --workers 4

# 线程池处理CPU密集型任务
executor = ThreadPoolExecutor(max_workers=10)

@app.get("/cpu-intensive")
async def cpu_intensive_task():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(executor, heavy_computation)
    return {"result": result}

# 并发控制
from asyncio import Semaphore

MAX_CONCURRENT = 100
semaphore = Semaphore(MAX_CONCURRENT)

@app.get("/limited-concurrent")
async def limited_concurrent():
    async with semaphore:
        result = await process_request()
        return result

# 连接池复用
import httpx

http_client = httpx.AsyncClient(
    limits=httpx.Limits(
        max_keepalive_connections=100,
        max_connections=200
    )
)

@app.get("/external-api")
async def call_external():
    response = await http_client.get("https://api.example.com/data")
    return response.json()

# 异步数据库连接池
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=20,
    max_overflow=10
)

# 响应流式处理
from fastapi.responses import StreamingResponse

@app.get("/stream")
async def stream_response():
    async def generate():
        for i in range(1000):
            yield f"data: {i}\n"
            await asyncio.sleep(0.01)
    
    return StreamingResponse(generate(), media_type="text/event-stream")

# 批量处理
@app.post("/batch")
async def batch_process(items: list):
    tasks = [process_item(item) for item in items]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return {"results": results}

10. FastAPI如何实现请求限流？

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import time
from collections import defaultdict
import asyncio

app = FastAPI()

# 令牌桶限流
class TokenBucket:
    def __init__(self, rate: int, capacity: int):
        self.rate = rate
        self.capacity = capacity
        self.buckets = {}
    
    def get_bucket(self, key: str):
        if key not in self.buckets:
            self.buckets[key] = {
                "tokens": self.capacity,
                "last_update": time.time()
            }
        return self.buckets[key]
    
    def is_allowed(self, key: str) -> bool:
        bucket = self.get_bucket(key)
        now = time.time()
        
        # 添加令牌
        elapsed = now - bucket["last_update"]
        bucket["tokens"] = min(
            self.capacity,
            bucket["tokens"] + elapsed * self.rate
        )
        bucket["last_update"] = now
        
        if bucket["tokens"] >= 1:
            bucket["tokens"] -= 1
            return True
        
        return False

limiter = TokenBucket(rate=10, capacity=100)

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    client_id = request.client.host
    
    if not limiter.is_allowed(client_id):
        return JSONResponse(
            status_code=429,
            content={"error": "Too many requests"}
        )
    
    return await call_next(request)

# 滑动窗口限流
class SlidingWindow:
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests = defaultdict(list)
    
    def is_allowed(self, key: str) -> bool:
        now = time.time()
        requests = self.requests[key]
        
        # 移除过期请求
        while requests and now - requests[0] > self.window_seconds:
            requests.pop(0)
        
        if len(requests) >= self.max_requests:
            return False
        
        requests.append(now)
        return True

# 分布式限流（Redis）
import redis.asyncio as redis

redis_client = redis.from_url("redis://localhost")

async def distributed_rate_limit(key: str, limit: int, window: int) -> bool:
    now = time.time()
    window_start = now - window
    
    pipe = redis_client.pipeline()
    await pipe.zremrangebyscore(key, 0, window_start)
    await pipe.zadd(key, {str(now): now})
    await pipe.zcard(key)
    await pipe.expire(key, window)
    
    results = await pipe.execute()
    return results[2] <= limit

@app.get("/limited")
async def limited_endpoint(request: Request):
    client_id = request.client.host
    
    if not await distributed_rate_limit(f"rate:{client_id}", 100, 60):
        raise HTTPException(status_code=429, detail="Too many requests")
    
    return {"message": "OK"}

11. FastAPI如何实现数据库连接池管理？

from fastapi import FastAPI, Depends
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

app = FastAPI()

# 同步连接池
sync_engine = create_engine(
    "postgresql://user:pass@localhost/db",
    pool_size=20,          # 连接池大小
    max_overflow=10,       # 最大溢出连接
    pool_timeout=30,       # 获取连接超时
    pool_recycle=3600,     # 连接回收时间
    pool_pre_ping=True,    # 连接前检查
    echo_pool=True         # 连接池日志
)

SyncSessionLocal = sessionmaker(bind=sync_engine)

# 异步连接池
async_engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=20,
    max_overflow=10,
    pool_timeout=30,
    pool_recycle=3600
)

AsyncSessionLocal = sessionmaker(
    async_engine,
    class_=AsyncSession,
    expire_on_commit=False
)

# 依赖注入
def get_sync_db():
    db = SyncSessionLocal()
    try:
        yield db
    finally:
        db.close()

async def get_async_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

# 使用连接池
@app.get("/users")
async def get_users(db: Session = Depends(get_sync_db)):
    return db.query(User).all()

@app.get("/async-users")
async def get_async_users(db: AsyncSession = Depends(get_async_db)):
    result = await db.execute(select(User))
    return result.scalars().all()

# 连接池监控
@app.get("/pool-stats")
async def pool_stats():
    return {
        "pool_size": sync_engine.pool.size(),
        "checked_in": sync_engine.pool.checkedin(),
        "checked_out": sync_engine.pool.checkedout(),
        "overflow": sync_engine.pool.overflow()
    }

# 连接池预热
@app.on_event("startup")
async def warmup_pool():
    # 预先创建连接
    for _ in range(10):
        db = SyncSessionLocal()
        db.execute("SELECT 1")
        db.close()

12. FastAPI如何实现缓存策略？

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import redis.asyncio as redis
from functools import wraps
import hashlib
import json

app = FastAPI()
redis_client = redis.from_url("redis://localhost")

# 缓存装饰器
def cache(ttl: int = 300):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # 生成缓存键
            cache_key = f"{func.__name__}:{hashlib.md5(str(args).encode()).hexdigest()}"
            
            # 尝试从缓存获取
            cached = await redis_client.get(cache_key)
            if cached:
                return json.loads(cached)
            
            # 执行函数
            result = await func(*args, **kwargs)
            
            # 存储到缓存
            await redis_client.setex(cache_key, ttl, json.dumps(result))
            
            return result
        return wrapper
    return decorator

# HTTP缓存头
@app.get("/items")
async def get_items(response: Response):
    items = fetch_items()
    
    # 设置缓存头
    response.headers["Cache-Control"] = "public, max-age=300"
    response.headers["ETag"] = compute_etag(items)
    
    return items

# 条件请求
@app.get("/conditional")
async def conditional_request(request: Request):
    data = get_data()
    etag = compute_etag(data)
    
    if request.headers.get("If-None-Match") == etag:
        return Response(status_code=304)
    
    return JSONResponse(content=data, headers={"ETag": etag})

# 多级缓存
class MultiLevelCache:
    def __init__(self):
        self.local_cache = {}
        self.redis = redis_client
    
    async def get(self, key: str):
        # L1: 本地缓存
        if key in self.local_cache:
            return self.local_cache[key]
        
        # L2: Redis缓存
        data = await self.redis.get(key)
        if data:
            self.local_cache[key] = json.loads(data)
            return self.local_cache[key]
        
        return None
    
    async def set(self, key: str, value, ttl: int = 300):
        self.local_cache[key] = value
        await self.redis.setex(key, ttl, json.dumps(value))
    
    async def delete(self, key: str):
        self.local_cache.pop(key, None)
        await self.redis.delete(key)

# 缓存失效
@app.post("/items")
async def create_item(item: ItemCreate):
    result = save_item(item)
    
    # 清除相关缓存
    await redis_client.delete("items_list")
    
    return result