中间件与生命周期与异常

import time
import uuid
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_process_time_and_trace_id(request: Request, call_next):
    # 1. 前置处理：生成/提取链路追踪 ID，记录起始时间戳
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    start_time = time.perf_counter()
    request.state.request_id = request_id

    # 2. 执行后续路由处理
    response = await call_next(request)

    # 3. 后置处理：计算接口耗时并写入响应头
    process_time = time.perf_counter() - start_time
    response.headers["X-Process-Time"] = f"{process_time:.4f}s"
    response.headers["X-Request-ID"] = request_id

    return response

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "http://localhost:3000",
    "http://localhost:5173",
    "https://your-production-app.com",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["X-Request-ID"],
    max_age=600
)

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
import asyncio

class AIModelManager:
    async def load(self):
        print("-> [Lifespan] 正在加载大模型权重至 GPU 显存...")
        await asyncio.sleep(0.5)

    async def cleanup(self):
        print("-> [Lifespan] 正在释放 GPU 资源与连接...")

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    ai_manager = AIModelManager()
    await ai_manager.load()
    app.state.ai_manager = ai_manager

    yield  # 服务在此对外提供 API 服务

    # Shutdown
    await ai_manager.cleanup()

app = FastAPI(lifespan=lifespan)

from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

app = FastAPI()

class BusinessException(Exception):
    def __init__(self, code: int, message: str):
        self.code = code
        self.message = message

@app.exception_handler(BusinessException)
async def business_exception_handler(request: Request, exc: BusinessException):
    return JSONResponse(
        status_code=status.HTTP_200_OK,
        content={"code": exc.code, "message": exc.message, "success": False}
    )

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = [{"field": " -> ".join(map(str, err["loc"])), "msg": err["msg"]} for err in exc.errors()]
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={"code": 42200, "message": "参数校验未通过", "errors": errors}
    )

from fastapi import FastAPI, BackgroundTasks
import asyncio

app = FastAPI()

async def send_welcome_email(email: str):
    await asyncio.sleep(1.0)
    print(f"-> [Email] 邮件已发送给: {email}")

@app.post("/register/")
async def register(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_welcome_email, email=email)
    return {"message": "注册成功，欢迎邮件正在后台发送"}