perf(api): share default error-response fields across routes

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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]:

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