FaqSearch/app/openapi.py

"""自定义 OpenAPI Schema 生成。

FastAPI 默认根据路由参数自动生成 OpenAPI 时，不会感知 exception_handlers，
导致文档里的错误响应（422 等）使用内置 HTTPValidationError 模板，
与项目实际返回的统一信封 {code, data, msg} 不一致。

本模块在应用初始化阶段覆写 app.openapi 函数，让所有错误响应文档与运行时返回保持一致。
"""
from __future__ import annotations

from typing import Any

from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi

from app.response import ResponseCode

# 需要在 OpenAPI 文档中替换的错误状态码 → (业务码, 描述, data 示例)
# 与 app/exception_handlers.py 中的处理器返回值保持一致
_ERROR_STATUS_MAP: dict[str, tuple[ResponseCode, str, Any]] = {
    "400": (
        ResponseCode.REQUEST_INVALID,
        "请求格式不正确",
        None,
    ),
    "404": (
        ResponseCode.NOT_FOUND,
        "资源不存在",
        None,
    ),
    "405": (
        ResponseCode.METHOD_NOT_ALLOWED,
        "请求方法不允许",
        None,
    ),
    "422": (
        ResponseCode.PARAM_INVALID,
        "参数验证失败",
        [
            {
                "type": "greater_than_equal",
                "loc": ["query", "page"],
                "msg": "Input should be greater than or equal to 1",
                "input": "-1",
                "ctx": {"ge": 1},
            }
        ],
    ),
    "500": (
        ResponseCode.INTERNAL_ERROR,
        "服务器内部错误",
        {"request_id": "uuid-for-tracing"},
    ),
    "502": (
        ResponseCode.UPSTREAM_ERROR,
        "上游服务调用失败（Zendesk 等）",
        None,
    ),
    "504": (
        ResponseCode.UPSTREAM_TIMEOUT,
        "上游服务响应超时",
        None,
    ),
}


def _build_error_schema_component() -> dict[str, Any]:
    """构造统一错误响应的 OpenAPI schema 组件。"""
    return {
        "type": "object",
        "properties": {
            "code": {
                "type": "integer",
                "description": "业务错误码（非 0 表示失败）",
                "example": ResponseCode.PARAM_INVALID.code,
            },
            "data": {
                "description": (
                    "失败时的附加数据：参数验证失败时为错误字段列表，"
                    "服务器内部错误时含 request_id，其他错误一般为 null"
                ),
                "nullable": True,
                "example": [
                    {
                        "type": "greater_than_equal",
                        "loc": ["query", "page"],
                        "msg": "Input should be greater than or equal to 1",
                        "input": "-1",
                        "ctx": {"ge": 1},
                    }
                ],
            },
            "msg": {
                "type": "string",
                "description": "错误描述",
                "example": ResponseCode.PARAM_INVALID.msg,
            },
        },
        "required": ["code", "msg"],
        "title": "ApiErrorResponse",
    }


def _build_response_for(
    code: ResponseCode, description: str, data_example: Any
) -> dict[str, Any]:
    """为某个具体错误状态码构造 OpenAPI 响应对象（带 example，便于 ReDoc 展示）。"""
    return {
        "description": f"{description}（业务码 {code.code}）",
        "content": {
            "application/json": {
                "schema": {"$ref": "#/components/schemas/ApiErrorResponse"},
                "example": {
                    "code": code.code,
                    "data": data_example,
                    "msg": code.msg,
                },
            }
        },
    }


def setup_custom_openapi(app: FastAPI) -> None:
    """覆写 app.openapi，让所有错误响应文档使用统一信封格式。

    应在应用初始化阶段调用一次（main.py 中 register_exception_handlers 之后）。
    """

    def custom_openapi() -> dict[str, Any]:
        if app.openapi_schema:
            return app.openapi_schema

        schema = get_openapi(
            title=app.title,
            version=app.version,
            description=app.description,
            routes=app.routes,
        )

        # 注册统一错误响应组件
        components = schema.setdefault("components", {}).setdefault("schemas", {})
        components["ApiErrorResponse"] = _build_error_schema_component()

        # 替换所有路径下的错误响应
        for path_item in schema.get("paths", {}).values():
            for operation in path_item.values():
                if not isinstance(operation, dict):
                    continue
                responses = operation.get("responses", {})
                for status_code, (
                    code,
                    description,
                    data_example,
                ) in _ERROR_STATUS_MAP.items():
                    if status_code in responses:
                        responses[status_code] = _build_response_for(
                            code, description, data_example
                        )

        app.openapi_schema = schema
        return schema

    app.openapi = custom_openapi  # type: ignore[method-assign]