From 08123fa5cff30f006e9f0f7595f3fc4574f420a4 Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Wed, 30 Sep 2026 17:36:17 +0800 Subject: [PATCH] perf(api): share default error-response fields across routes Co-Authored-By: Claude Opus 5.5 --- app/api/response.py | 41 +++++++++++++++++++++++-- tests/test_api_response.py | 63 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 102 insertions(+), 2 deletions(-) diff --git a/app/api/response.py b/app/api/response.py index aad8a61ea..aecfb79dc 100644 --- a/app/api/response.py +++ b/app/api/response.py @@ -7,6 +7,7 @@ from fastapi import APIRouter, Query, Request from fastapi.datastructures import DefaultPlaceholder from fastapi.responses import JSONResponse from fastapi.routing import APIRoute, get_typed_return_annotation +from fastapi.utils import ModelField, create_model_field from starlette.responses import Response as StarletteResponse from app.schemas.common import JsonData @@ -24,6 +25,19 @@ ERROR_RESPONSES: dict[int, dict[str, Any]] = { }, 500: {"model": Response[None], "description": "服务器内部错误"}, } +# 默认错误响应只用于生成 OpenAPI 文档。FastAPI 会为每个路由的每个带 model 的附加响应单独创建 +# ModelField，并各自编译一套 pydantic 校验器与序列化器；全部路由改为共用下面这一组字段，文档输出不变。 +# 构造路由时先传入去掉 model 的声明，使 FastAPI 不再逐路由创建字段，构造后再换回原声明并挂上共享字段。 +_ERROR_RESPONSES_WITHOUT_MODEL: dict[int, dict[str, Any]] = { + status: {key: value for key, value in spec.items() if key != "model"} + for status, spec in ERROR_RESPONSES.items() +} +_SHARED_ERROR_RESPONSE_FIELDS: dict[int, ModelField] = { + status: create_model_field( + name=f"Response_{status}", type_=spec["model"], mode="serialization" + ) + for status, spec in ERROR_RESPONSES.items() +} RAW_RESPONSE_OPENAPI_KEY = "x-moviepilot-raw-response" COLLECTION_PAGINATION_OPENAPI_KEY = "x-moviepilot-compatible-pagination" COLLECTION_TOTAL_OPENAPI_KEY = "x-moviepilot-exact-total" @@ -167,6 +181,7 @@ class ResponseAPIRoute(APIRoute): ) super().__init__(path=path, endpoint=endpoint, **kwargs) + self._attach_shared_error_response_fields() def get_route_handler( self, @@ -371,10 +386,32 @@ class ResponseAPIRoute(APIRoute): def _merge_error_responses( responses: dict[int | str, dict[str, Any]] | None, ) -> dict[int | str, dict[str, Any]]: - """补齐统一错误模型，并保留端点已经显式声明的响应。""" + """补齐统一错误模型，并保留端点已经显式声明的响应。 + + 仍是默认声明的状态码换成不带 model 的副本，交给 ``_attach_shared_error_response_fields`` + 在路由构造后接上共享字段；端点显式覆盖的响应原样保留，由 FastAPI 按原逻辑建字段。 + """ merged_responses: dict[int | str, dict[str, Any]] = dict(ERROR_RESPONSES) merged_responses.update(responses or {}) - return merged_responses + return { + status: ( + _ERROR_RESPONSES_WITHOUT_MODEL[status] + if isinstance(status, int) and spec is ERROR_RESPONSES.get(status) + else spec + ) + for status, spec in merged_responses.items() + } + + def _attach_shared_error_response_fields(self) -> None: + """把默认错误响应换回原声明并挂上共享字段。 + + ``responses`` 保留完整声明，``include_router`` 用它重建路由时按对象身份再次走共享路径； + FastAPI 生成 OpenAPI 时只从 ``response_fields`` 取 schema，字段名不进入输出。 + """ + for status, spec in ERROR_RESPONSES.items(): + if self.responses.get(status) is _ERROR_RESPONSES_WITHOUT_MODEL[status]: + self.responses[status] = spec + self.response_fields[status] = _SHARED_ERROR_RESPONSE_FIELDS[status] @staticmethod def _wrap_endpoint(endpoint: Callable[..., Any]) -> Callable[..., Any]: diff --git a/tests/test_api_response.py b/tests/test_api_response.py index 5fe13b2ce..2b006ab46 100644 --- a/tests/test_api_response.py +++ b/tests/test_api_response.py @@ -21,6 +21,7 @@ from app.adapters.web.plugin.routes import FastAPIDynamicRouteRegistry from app.api.response import ( COLLECTION_PAGINATION_OPENAPI_KEY, COLLECTION_TOTAL_OPENAPI_KEY, + ERROR_RESPONSES, RAW_RESPONSE_OPENAPI_KEY, CompatibleCountParam, CompatiblePageParam, @@ -925,6 +926,68 @@ def test_all_openapi_error_responses_use_json_schemas(): assert invalid_responses == [] +def test_host_routes_share_default_error_response_fields(): + """全部统一响应路由的默认错误响应共用同一组字段，路由声明仍保留完整的错误模型。""" + routes = [ + route + for _, route in _v1_compat_routes() + if isinstance(route, ResponseAPIRoute) + ] + assert len(routes) > 100 + + for status, spec in ERROR_RESPONSES.items(): + default_routes = [ + route for route in routes if route.responses.get(status) is spec + ] + assert default_routes, status + assert len({id(route.response_fields[status]) for route in default_routes}) == 1 + + +class _PerRouteErrorResponseRoute(ResponseAPIRoute): + """对照组：按 FastAPI 原生方式为每个路由单独创建错误响应字段。""" + + @staticmethod + def _merge_error_responses(responses): + merged = dict(ERROR_RESPONSES) + merged.update(responses or {}) + return merged + + def _attach_shared_error_response_fields(self) -> None: + pass + + +def _error_response_openapi(route_class: type[APIRoute]) -> dict[str, Any]: + """构造包含默认错误响应和端点自定义 404 的最小应用并返回 OpenAPI。""" + app = FastAPI() + app.router.route_class = route_class + router = APIRouter(route_class=route_class) + + @router.get("/items", response_model=list[Item]) + async def list_items() -> list[Item]: + return [] + + @router.get( + "/items/{item_id}", + response_model=Item, + responses={404: {"model": Item, "description": "自定义缺失"}}, + ) + async def get_item(item_id: int) -> Item: + return Item(id=item_id) + + app.include_router(router, prefix="/v1") + return app.openapi() + + +def test_shared_error_response_fields_keep_openapi_identical(): + """共用错误响应字段后生成的 OpenAPI 与逐路由建字段完全一致，端点自定义响应不受影响。""" + shared = _error_response_openapi(ResponseAPIRoute) + per_route = _error_response_openapi(_PerRouteErrorResponseRoute) + + assert shared == per_route + not_found = shared["paths"]["/v1/items/{item_id}"]["get"]["responses"]["404"] + assert not_found["content"]["application/json"]["schema"]["$ref"].endswith("/Item") + + def test_openapi_success_models_have_no_implicit_empty_nested_schemas(): """2xx 响应可达模型不得包含裸 Any、裸数组或未声明值类型的开放映射。""" from app.factory import create_app