patbond-ai/app/api/main.py

"""FastAPI 应用工厂。

装配顺序：读配置 → 建适配器注册表 → 挂路由与异常处理器。配置与注册表放在
`app.state` 上，请求处理时从 `request.app` 取——**不用模块级全局**，这样测试能起
多个配置不同的实例而互不干扰。

**异常处理器是本模块的重点**：契约定义了两类错误信封（`kind = service` /
`provider`），而 FastAPI 默认的校验错误是 `{"detail": [...]}`，与两者都不符。若不
接管，调用方就要处理三种形状——而第三种还是框架的实现细节。
"""

from __future__ import annotations

import logging

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

from app.assembly import build_registry
from app.config import Settings

from .errors import ProviderError, ServiceError, ServiceErrorCode
from .routes import router

log = logging.getLogger(__name__)


def create_app(settings: Settings | None = None) -> FastAPI:
    """建应用。

    :param settings: 省略即从环境变量读。测试传入显式配置以免依赖环境。
    """
    resolved = settings or Settings.from_env()
    logging.basicConfig(level=resolved.log_level)

    app = FastAPI(
        title="patbond-ai 内部接口",
        version="0.1.0",
        description=(
            "patbond-creation（Java）调用本服务执行 AI 生成的内部契约。"
            "权威契约见仓库根目录的 openapi-internal.yaml——"
            "本服务自动生成的 /openapi.json 仅供调试，不是契约。"
        ),
        # 内部服务不开交互文档：少一个对外表面，且契约以手写 yaml 为准。
        docs_url=None,
        redoc_url=None,
    )
    app.state.settings = resolved
    app.state.registry = build_registry(resolved)

    app.include_router(router)
    _install_exception_handlers(app)
    _install_health(app)
    return app


def _install_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(ServiceError)
    def handle_service_error(request: Request, exc: ServiceError) -> JSONResponse:
        return JSONResponse(status_code=exc.code.http_status, content=exc.to_envelope())

    @app.exception_handler(ProviderError)
    def handle_provider_error(request: Request, exc: ProviderError) -> JSONResponse:
        log.warning(
            "服务商调用失败 code=%s retry=%s",
            exc.failure.provider_code,
            exc.failure.retry.value,
        )
        return JSONResponse(
            status_code=ProviderError.HTTP_STATUS, content=exc.to_envelope()
        )

    @app.exception_handler(RequestValidationError)
    def handle_validation_error(
        request: Request, exc: RequestValidationError
    ) -> JSONResponse:
        """把 FastAPI 的校验错误翻成 service 信封。

        默认形状是 `{"detail": [...]}`，与契约的两类信封都不符。不接管的话调用方
        要处理三种形状，而第三种还是框架的实现细节——换个框架就变了。
        """
        first = exc.errors()[0] if exc.errors() else {}
        # 字段路径形如 ("body", "duration_sec")，去掉 "body" 前缀更可读。
        location = ".".join(
            str(part) for part in first.get("loc", ()) if part != "body"
        )
        detail = first.get("msg", "请求体不合契约")
        message = f"{location}: {detail}" if location else detail
        error = ServiceError(ServiceErrorCode.INVALID_REQUEST, message)
        return JSONResponse(
            status_code=error.code.http_status, content=error.to_envelope()
        )

    @app.exception_handler(Exception)
    def handle_unexpected(request: Request, exc: Exception) -> JSONResponse:
        """兜底：本服务的 bug。

        完整堆栈只进日志，绝不回给调用方——内部细节外泄没有任何好处。
        """
        log.exception("未处理的异常")
        error = ServiceError(ServiceErrorCode.INTERNAL_ERROR, "服务内部错误")
        return JSONResponse(
            status_code=error.code.http_status, content=error.to_envelope()
        )


def _install_health(app: FastAPI) -> None:
    @app.get("/health", include_in_schema=False)
    def health() -> dict:
        """探活。**在 /internal 之外，故不需要服务票**——与 Java 侧各服务
        `/health` 不挂鉴权的既有约定一致。

        同时报出已装配的适配器：缺凭证时适配器不装配，而那个症状（任务永远排队）
        离根因很远，让探活直接说出来能省很多排查时间。
        """
        registry = app.state.registry
        return {
            "status": "ok",
            "providers": sorted(registry.codes),
        }