"""统一错误体与领域异常(契约见 docs/contracts/api-conventions.md 第 7 节)。 `code` 是稳定机器值;`message` 可本地化;`details` 不稳定但必须脱敏。 """ from typing import Any from fastapi import FastAPI, Request from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse from pydantic import BaseModel class ErrorBody(BaseModel): code: str message: str trace_id: str | None = None request_id: str | None = None details: dict[str, Any] | None = None class ErrorEnvelope(BaseModel): error: ErrorBody class HunjingError(Exception): """领域异常基类:映射为稳定错误码的 HTTP 响应。""" status_code: int = 500 code: str = "internal" def __init__(self, message: str, *, details: dict[str, Any] | None = None) -> None: super().__init__(message) self.message = message self.details = details class ValidationFailed(HunjingError): status_code = 422 code = "validation_failed" def _envelope( code: str, message: str, request: Request, details: dict[str, Any] | None ) -> dict[str, Any]: return { "error": { "code": code, "message": message, "trace_id": None, "request_id": request.headers.get("Hunjing-Request-ID"), "details": details, } } def install_error_handlers(app: FastAPI) -> None: @app.exception_handler(HunjingError) async def _hunjing_error(request: Request, exc: HunjingError) -> JSONResponse: return JSONResponse( status_code=exc.status_code, content=_envelope(exc.code, exc.message, request, exc.details), ) @app.exception_handler(RequestValidationError) async def _validation_error(request: Request, exc: RequestValidationError) -> JSONResponse: field_errors = [ {"path": ".".join(str(p) for p in err["loc"]), "reason": err["type"]} for err in exc.errors() ] return JSONResponse( status_code=422, content=_envelope( "validation_failed", "请求字段不合法。", request, {"field_errors": field_errors} ), )