API 版本管理三大核心实践：兼容旧版、平滑升级与灰度切流

from fastapi import FastAPI, APIRouter, Header, HTTPException
from pydantic import BaseModel
from typing import Optional, Dict, Any

app = FastAPI(title="User API")

# ========== 数据模型 ==========
class UserV1(BaseModel):
    id: int
    name: str

class UserV2(BaseModel):
    id: int
    username: str
    email: Optional[str] = None
    status: int = 1

# ========== 业务逻辑层 ==========
def get_user_from_db(user_id: int) -> Dict[str, Any]:
    """模拟数据库查询，返回最新结构数据"""
    return {
        "id": user_id,
        "username": "zhangsan",
        "email": "zhangsan@example.com",
        "status": 1
    }

# ========== 版本适配层 ==========
def adapt_to_v1(data: Dict[str, Any]) -> Dict[str, Any]:
    """新版数据结构 -> v1 结构适配"""
    return {
        "id": data["id"],
        "name": data["username"]  # 字段映射：username -> name
    }

def adapt_to_v2(data: Dict[str, Any]) -> Dict[str, Any]:
    """新版数据直接返回"""
    return data

# ========== 路由分发 ==========
v1_router = APIRouter(prefix="/api/v1", tags=["v1"])
v2_router = APIRouter(prefix="/api/v2", tags=["v2"])

@v1_router.get("/users/{user_id}", response_model=UserV1)
def get_user_v1(user_id: int):
    raw = get_user_from_db(user_id)
    return adapt_to_v1(raw)

@v2_router.get("/users/{user_id}", response_model=UserV2)
def get_user_v2(user_id: int):
    raw = get_user_from_db(user_id)
    return adapt_to_v2(raw)

app.include_router(v1_router)
app.include_router(v2_router)

import functools
from datetime import datetime
from fastapi import Response
import warnings

def deprecated(sunset_date: str, new_endpoint: str):
    """
    接口废弃装饰器
    :param sunset_date: 下线日期 YYYY-MM-DD
    :param new_endpoint: 新版接口路径
    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 提取 response 对象（FastAPI 依赖注入）
            response = kwargs.get("response")
            if response and isinstance(response, Response):
                response.headers["Deprecation"] = "true"
                response.headers["Sunset"] = sunset_date
                response.headers["Link"] = f'<{new_endpoint}>; rel="successor-version"'
            
            # 服务端日志告警
            warnings.warn(
                f"Endpoint {func.__name__} is deprecated, "
                f"will be removed on {sunset_date}. "
                f"Use {new_endpoint} instead.",
                DeprecationWarning,
                stacklevel=2
            )
            return func(*args, **kwargs)
        return wrapper
    return decorator

# 使用示例
@v1_router.get("/users/{user_id}")
@deprecated(sunset_date="2024-12-31", new_endpoint="/api/v2/users/{user_id}")
def get_user_v1(user_id: int, response: Response):
    raw = get_user_from_db(user_id)
    return adapt_to_v1(raw)

import hashlib
import random
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse

class GrayReleaseMiddleware(BaseHTTPMiddleware):
    """
    灰度发布中间件
    支持：比例切流、用户ID白名单、自定义Header灰度
    """
    
    def __init__(self, app, gray_config: dict):
        super().__init__(app)
        self.gray_percent = gray_config.get("percent", 0)  # 0-100
        self.gray_user_ids = set(gray_config.get("user_ids", []))
        self.gray_header = gray_config.get("header_key", "X-Gray-Version")
    
    def _is_gray_by_percent(self, unique_key: str) -> bool:
        """基于哈希的比例灰度，保证同一用户结果稳定"""
        if self.gray_percent <= 0:
            return False
        if self.gray_percent >= 100:
            return True
        hash_val = int(hashlib.md5(unique_key.encode()).hexdigest(), 16)
        return (hash_val % 100) < self.gray_percent
    
    def _get_unique_key(self, request: Request) -> str:
        """提取灰度分流的唯一标识，优先用户ID，其次IP"""
        user_id = request.headers.get("X-User-ID")
        if user_id:
            return f"user:{user_id}"
        client_ip = request.client.host if request.client else "unknown"
        return f"ip:{client_ip}"
    
    async def dispatch(self, request: Request, call_next):
        # 1. 强制灰度 Header 优先级最高
        gray_version = request.headers.get(self.gray_header)
        if gray_version:
            request.state.api_version = gray_version
            return await call_next(request)
        
        # 2. 用户白名单灰度
        user_id = request.headers.get("X-User-ID")
        if user_id and user_id in self.gray_user_ids:
            request.state.api_version = "v2"
            return await call_next(request)
        
        # 3. 比例灰度
        unique_key = self._get_unique_key(request)
        if self._is_gray_by_percent(unique_key):
            request.state.api_version = "v2"
        else:
            request.state.api_version = "v1"
        
        response = await call_next(request)
        # 响应头回注实际命中版本，便于排查
        response.headers["X-Api-Version"] = request.state.api_version
        return response

# 注册中间件
gray_config = {
    "percent": 10,          # 10% 流量切到新版
    "user_ids": ["1001", "1002", "admin"],  # 白名单用户强制新版
    "header_key": "X-Gray-Version"
}

app.add_middleware(GrayReleaseMiddleware, gray_config=gray_config)