FaqSearch/app/core/response.py

"""统一响应类与业务错误码定义。

所有接口响应统一使用 `{code, data, msg}` 信封格式：
    - code: 0 表示成功；其他值代表不同的失败类型
    - data: 业务数据（成功时）或 None（失败时）
    - msg:  描述信息

错误码使用 `(code, default_msg)` 的元组定义，code 和默认提示一一对应。
"""
from __future__ import annotations

from enum import Enum
from typing import Any, Generic, TypeVar

from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

T = TypeVar("T")


class ResponseCode(Enum):
    """业务错误码枚举：每个成员都是 (code, default_msg) 元组。

    分段规则：
        0          成功
        1000-1999  通用错误（参数、请求格式等）
        2000-2999  业务逻辑错误（缓存未就绪、资源不存在等）
        3000-3999  外部依赖错误（Zendesk、上游服务等）
        9000-9999  服务器内部错误

    用法：
        ResponseCode.SUCCESS.code   # -> 0
        ResponseCode.SUCCESS.msg    # -> "成功"
    """

    # 成功
    SUCCESS = (0, "成功")

    # 通用错误 1xxx
    PARAM_INVALID = (1001, "参数验证失败")
    REQUEST_INVALID = (1002, "请求格式不正确")
    METHOD_NOT_ALLOWED = (1003, "请求方法不允许")
    NOT_FOUND = (1004, "资源不存在")

    # 业务错误 2xxx
    CACHE_NOT_READY = (2001, "FAQ 缓存未就绪，请稍后重试")
    RESOURCE_NOT_FOUND = (2002, "业务资源不存在")

    # 外部依赖错误 3xxx
    UPSTREAM_ERROR = (3001, "上游服务调用失败")
    UPSTREAM_TIMEOUT = (3002, "上游服务响应超时")

    # 服务器错误 9xxx
    INTERNAL_ERROR = (9999, "服务器内部错误，请稍后重试")

    def __init__(self, code: int, msg: str) -> None:
        self.code = code
        self.msg = msg

    def __repr__(self) -> str:  # pragma: no cover
        return f"<{self.__class__.__name__}.{self.name}: code={self.code} msg={self.msg!r}>"


class ApiResponse(BaseModel, Generic[T]):
    """统一 API 响应模型。

    用于在路由 `response_model` 中声明，让 OpenAPI 文档展示标准结构。

    用法示例：
        @router.get("/search", response_model=ApiResponse[SearchData])
        async def search(...) -> ApiResponse[SearchData]:
            return ApiResponse.ok(data=SearchData(...))

        # 失败响应（自动使用错误码对应的默认 msg）
        return ApiResponse.fail(ResponseCode.PARAM_INVALID)

        # 失败响应（覆盖默认 msg）
        return ApiResponse.fail(ResponseCode.UPSTREAM_ERROR, msg="Zendesk 限流")
    """

    code: int = Field(default=0, description="业务码，0=成功，非0=失败")
    data: T | None = Field(default=None, description="业务数据，失败时为 null")
    msg: str = Field(default="成功", description="描述信息")

    @classmethod
    def ok(
        cls,
        data: T | None = None,
        msg: str | None = None,
    ) -> "ApiResponse[T]":
        """构造成功响应。msg 为空时使用 SUCCESS 默认提示。"""
        return cls(
            code=ResponseCode.SUCCESS.code,
            data=data,
            msg=msg if msg is not None else ResponseCode.SUCCESS.msg,
        )

    @classmethod
    def fail(
        cls,
        code: ResponseCode,
        msg: str | None = None,
        data: T | None = None,
    ) -> "ApiResponse[T]":
        """构造失败响应。msg 为空时使用错误码自带的默认提示。"""
        return cls(
            code=code.code,
            data=data,
            msg=msg if msg is not None else code.msg,
        )


def success_response(
    data: Any = None,
    msg: str | None = None,
    http_status: int = 200,
) -> JSONResponse:
    """构造成功的 JSONResponse（适用于异常处理器、中间件等场景）。"""
    return JSONResponse(
        status_code=http_status,
        content={
            "code": ResponseCode.SUCCESS.code,
            "data": data,
            "msg": msg if msg is not None else ResponseCode.SUCCESS.msg,
        },
    )


def error_response(
    code: ResponseCode,
    msg: str | None = None,
    data: Any = None,
    http_status: int = 200,
    headers: dict[str, str] | None = None,
) -> JSONResponse:
    """构造失败的 JSONResponse（适用于异常处理器、中间件等场景）。

    Args:
        code: 业务错误码枚举
        msg:  错误提示，为空时使用错误码自带的默认提示
        data: 附加数据（如参数验证错误的详细字段列表）
        http_status: HTTP 状态码，默认 200（业务错误也走 200，由 code 区分成败）
        headers: 额外响应头（如 401 必须带 WWW-Authenticate 才能让浏览器弹登录框）

    返回示例：
        {"code": 1001, "data": [...], "msg": "参数验证失败"}
    """
    return JSONResponse(
        status_code=http_status,
        content={
            "code": code.code,
            "data": data,
            "msg": msg if msg is not None else code.msg,
        },
        headers=headers,
    )


class BusinessError(Exception):
    """业务异常基类，用于在业务层抛出后由全局处理器统一转为 ApiResponse。

    用法：
        raise BusinessError(ResponseCode.CACHE_NOT_READY)
        raise BusinessError(ResponseCode.UPSTREAM_ERROR, msg="Zendesk 限流")
    """

    def __init__(
        self,
        code: ResponseCode,
        msg: str | None = None,
        data: Any = None,
        http_status: int = 200,
    ) -> None:
        self.response_code = code
        self.code = code.code
        self.msg = msg if msg is not None else code.msg
        self.data = data
        self.http_status = http_status
        super().__init__(self.msg)