xiaoxia-saas/apps/api/app/middleware/versioning.py

"""
API 版本管理中间件
"""

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware


class APIVersionMiddleware(BaseHTTPMiddleware):
    """API 版本管理中间件"""

    # 版本配置
    VERSIONS = {
        "v1": {
            "status": "stable",
            "deprecated": False,
            "sunset_date": None,
            "release_date": "2026-06-17",
        },
        "v2": {
            "status": "development",
            "deprecated": False,
            "sunset_date": None,
            "release_date": None,
        },
    }

    async def dispatch(self, request: Request, call_next):
        # 提取版本号
        version = self._extract_version(request.url.path)

        # 处理请求
        response = await call_next(request)

        # 添加版本信息头
        if version:
            response.headers["X-API-Version"] = version

            # 添加弃用警告
            version_info = self.VERSIONS.get(version, {})
            if version_info.get("deprecated"):
                response.headers["X-API-Deprecated"] = "true"

                sunset_date = version_info.get("sunset_date")
                if sunset_date:
                    response.headers["X-API-Sunset-Date"] = sunset_date

                response.headers["X-API-Deprecation-Info"] = f"https://docs.xiaoxia-saas.com/api/deprecation/{version}"

        return response

    def _extract_version(self, path: str) -> str:
        """从路径中提取版本号"""
        parts = path.split("/")
        for part in parts:
            if part.startswith("v") and part[1:].isdigit():
                return part
        return None


class VersionNotFoundMiddleware(BaseHTTPMiddleware):
    """处理已下线的 API 版本"""

    SUNSET_VERSIONS: list[str] = []  # 已下线的版本列表

    async def dispatch(self, request: Request, call_next):
        version = self._extract_version(request.url.path)

        if version in self.SUNSET_VERSIONS:
            from fastapi.responses import JSONResponse

            return JSONResponse(
                status_code=410,
                content={
                    "error": {
                        "code": "API_VERSION_SUNSET",
                        "message": f"API {version} has been sunset and is no longer available",
                        "sunset_date": "2028-07-01",
                        "migration_guide": f"https://docs.xiaoxia-saas.com/api/migration/{version}",
                    }
                },
            )

        return await call_next(request)

    def _extract_version(self, path: str) -> str:
        """从路径中提取版本号"""
        parts = path.split("/")
        for part in parts:
            if part.startswith("v") and part[1:].isdigit():
                return part
        return None