From 03c9ab33b1bd3e03b46b6f32f80f138370a56050 Mon Sep 17 00:00:00 2001 From: Lev Vereshchagin Date: Thu, 22 Oct 2026 23:02:12 +0300 Subject: [PATCH 1/2] Add configurable HTTP error titles to web bootstrappers --- README.md | 18 ++ microbootstrap/__init__.py | 2 + microbootstrap/bootstrappers/fastapi.py | 63 ++++ microbootstrap/bootstrappers/litestar.py | 76 +++++ .../http_error_titles_instrument.py | 27 ++ microbootstrap/settings.py | 3 + tests/bootstrappers/test_http_error_titles.py | 303 ++++++++++++++++++ 7 files changed, 492 insertions(+) create mode 100644 microbootstrap/instruments/http_error_titles_instrument.py create mode 100644 tests/bootstrappers/test_http_error_titles.py diff --git a/README.md b/README.md index ad80697..f963561 100644 --- a/README.md +++ b/README.md @@ -751,6 +751,24 @@ Application-specific descriptions and authentication declarations remain with th application, using native framework metadata and the existing `security_schemes` setting. No authentication policy is inferred or enforced by this option. +### HTTP error titles + +Enable `http_error_titles_enabled=True` in `FastApiSettings`, `LitestarSettings`, +or `HttpErrorTitlesConfig` to add the HTTP status phrase as `title` on default +JSON error responses. The option is disabled by default; the normal settings +environment prefix applies to `HTTP_ERROR_TITLES_ENABLED`. + +The native error fields, status codes and headers are preserved. Default Litestar +errors use JSON even when a route's success media type is different. An unknown +HTTP status receives the title `HTTP error`. Existing application exception +handlers keep precedence and remain responsible for their own payloads and +schemas. Explicit error responses returned by application handlers are also +owned by the application. + +The bootstrappers document `title` as required on their generated validation +schemas when the default handler is active. This is independent of +`openapi_descriptions_enabled`. Debug traceback responses remain framework-owned. + ## Configuration While settings provide a convenient mechanism, it's not always feasible to store everything within them. diff --git a/microbootstrap/__init__.py b/microbootstrap/__init__.py index 51fe81e..cc53b63 100644 --- a/microbootstrap/__init__.py +++ b/microbootstrap/__init__.py @@ -1,5 +1,6 @@ from microbootstrap.instruments.cors_instrument import CorsConfig from microbootstrap.instruments.health_checks_instrument import HealthChecksConfig +from microbootstrap.instruments.http_error_titles_instrument import HttpErrorTitlesConfig from microbootstrap.instruments.logging_instrument import LoggingConfig from microbootstrap.instruments.openapi_security_schemes import ( OpenApiApiKeySecurityScheme, @@ -50,6 +51,7 @@ "FastStreamSettings", "FastStreamTelemetryMiddlewareProtocol", "HealthChecksConfig", + "HttpErrorTitlesConfig", "InstrumentsSetupperSettings", "LitestarPrometheusConfig", "LitestarSettings", diff --git a/microbootstrap/bootstrappers/fastapi.py b/microbootstrap/bootstrappers/fastapi.py index fd54101..d5fb519 100644 --- a/microbootstrap/bootstrappers/fastapi.py +++ b/microbootstrap/bootstrappers/fastapi.py @@ -1,16 +1,22 @@ import contextlib +import http import typing import fastapi +from fastapi.encoders import jsonable_encoder +from fastapi.exceptions import RequestValidationError from fastapi.middleware.cors import CORSMiddleware from fastapi_offline_docs import enable_offline_docs from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor from prometheus_fastapi_instrumentator import Instrumentator, metrics +from starlette.exceptions import HTTPException +from starlette.responses import JSONResponse from microbootstrap.bootstrappers.base import ApplicationBootstrapper from microbootstrap.config.fastapi import FastApiConfig from microbootstrap.instruments.cors_instrument import CorsInstrument from microbootstrap.instruments.health_checks_instrument import HealthChecksInstrument, HealthCheckTypedDict +from microbootstrap.instruments.http_error_titles_instrument import HttpErrorTitlesInstrument from microbootstrap.instruments.logging_instrument import LoggingInstrument from microbootstrap.instruments.openapi_security_schemes import serialize_security_schemes from microbootstrap.instruments.openapi_version_docs import SUPPORTED_HTTP_METHODS @@ -144,6 +150,63 @@ def _document_operations(self, openapi_schema: dict[str, typing.Any]) -> None: operation["x-accept-versioning"], operation["description"] = documentation +@FastApiBootstrapper.use_instrument() +class FastApiHttpErrorTitlesInstrument(HttpErrorTitlesInstrument): + async def http_exception_response(self, request: fastapi.Request, exc: HTTPException) -> fastapi.Response: + response = await fastapi.exception_handlers.http_exception_handler(request, exc) + if exc.status_code >= http.HTTPStatus.BAD_REQUEST: + return JSONResponse( + {"detail": exc.detail, "title": self.title_for_status(exc.status_code)}, + status_code=exc.status_code, + headers=exc.headers, + ) + return response + + async def validation_error_response(self, _request: fastapi.Request, exc: RequestValidationError) -> JSONResponse: + return JSONResponse( + { + "detail": jsonable_encoder(exc.errors()), + "title": self.title_for_status(http.HTTPStatus.UNPROCESSABLE_ENTITY), + }, + status_code=http.HTTPStatus.UNPROCESSABLE_ENTITY, + ) + + async def internal_error_response(self, _request: fastapi.Request, _exc: Exception) -> JSONResponse: + return JSONResponse( + {"detail": "Internal Server Error", "title": self.title_for_status(http.HTTPStatus.INTERNAL_SERVER_ERROR)}, + status_code=http.HTTPStatus.INTERNAL_SERVER_ERROR, + ) + + def bootstrap_before(self) -> dict[str, typing.Any]: + return { + "exception_handlers": { + HTTPException: self.http_exception_response, + RequestValidationError: self.validation_error_response, + Exception: self.internal_error_response, + } + } + + def bootstrap_after(self, application: ApplicationT) -> ApplicationT: + if application.exception_handlers.get(RequestValidationError) != self.validation_error_response: + return application + original_openapi: typing.Final = application.openapi + + def titled_openapi() -> dict[str, typing.Any]: + document: typing.Final = original_openapi() + schema = document.get("components", {}).get("schemas", {}).get("HTTPValidationError") + if schema is not None: + schema.setdefault("properties", {}).setdefault( + "title", {"type": "string", "description": "Short, human-readable summary of the HTTP error."} + ) + required = schema.setdefault("required", []) + if "title" not in required: + required.append("title") + return document + + application.openapi = titled_openapi # type: ignore[method-assign] # FastAPI's public custom OpenAPI hook. + return application + + @FastApiBootstrapper.use_instrument() class FastApiCorsInstrument(CorsInstrument): def bootstrap_after(self, application: ApplicationT) -> ApplicationT: diff --git a/microbootstrap/bootstrappers/litestar.py b/microbootstrap/bootstrappers/litestar.py index df77906..de3277e 100644 --- a/microbootstrap/bootstrappers/litestar.py +++ b/microbootstrap/bootstrappers/litestar.py @@ -1,5 +1,7 @@ from __future__ import annotations import dataclasses +import http +import json import typing import litestar @@ -11,8 +13,11 @@ OpenTelemetryConfig as LitestarOpentelemetryConfig, ) from litestar.contrib.prometheus import PrometheusConfig, PrometheusController +from litestar.exceptions import HTTPException +from litestar.exceptions.responses import create_debug_response, create_exception_response from litestar.middleware import ASGIMiddleware from litestar.openapi.plugins import SwaggerRenderPlugin +from litestar.routes import HTTPRoute from litestar.types.asgi_types import ASGIApp, Scope from litestar_offline_docs import generate_static_files_config from opentelemetry.instrumentation.asgi import OpenTelemetryMiddleware @@ -27,6 +32,7 @@ HealthChecksInstrument, HealthCheckTypedDict, ) +from microbootstrap.instruments.http_error_titles_instrument import HttpErrorTitlesInstrument from microbootstrap.instruments.logging_instrument import LoggingInstrument from microbootstrap.instruments.openapi_security_schemes import ( OpenApiApiKeySecurityScheme, @@ -278,6 +284,76 @@ def _build_litestar_security_scheme(cls, security_scheme: _OpenApiSecurityScheme return openapi.spec.SecurityScheme(**scheme_data) +@LitestarBootstrapper.use_instrument() +class LitestarHttpErrorTitlesInstrument(HttpErrorTitlesInstrument): + def exception_response( + self, request: litestar.Request[typing.Any, typing.Any, typing.Any], exc: Exception + ) -> litestar.Response[typing.Any]: + status_code = exc.status_code if isinstance(exc, HTTPException) else http.HTTPStatus.INTERNAL_SERVER_ERROR + if status_code == http.HTTPStatus.INTERNAL_SERVER_ERROR and request.app.debug: + return create_debug_response(request, exc) + response = create_exception_response(request, exc) + if status_code >= http.HTTPStatus.BAD_REQUEST: + if not isinstance(response.content, dict): + response.content = json.loads(response.content) + response.content.setdefault("title", self.title_for_status(status_code)) + response.media_type = litestar.MediaType.JSON + return response + + def bootstrap_before(self) -> dict[str, typing.Any]: + return {"exception_handlers": {Exception: self.exception_response}} + + @classmethod + def _title_error_schema(cls, schema: openapi.spec.Schema) -> None: + for branch in (*(schema.one_of or []), *(schema.any_of or []), *(schema.all_of or [])): + if isinstance(branch, openapi.spec.Schema): + cls._title_error_schema(branch) + properties = schema.properties or {} + if "status_code" in properties and "detail" in properties: + properties.setdefault( + "title", + openapi.spec.Schema( + type=openapi.spec.OpenAPIType.STRING, + description="Short, human-readable summary of the HTTP error.", + ), + ) + schema.required = list(dict.fromkeys([*(schema.required or []), "title"])) + for example in schema.examples or []: + if isinstance(example, dict) and isinstance(example.get("status_code"), int): + example.setdefault("title", cls.title_for_status(example["status_code"])) + + @classmethod + def _title_operation(cls, operation: openapi.spec.Operation) -> None: + for code, response in (operation.responses or {}).items(): + if str(code).startswith(("4", "5")) and isinstance(response, openapi.spec.OpenAPIResponse): + for media_type in (response.content or {}).values(): + if isinstance(media_type.schema, openapi.spec.Schema): + cls._title_error_schema(media_type.schema) + + def bootstrap_after(self, application: ApplicationT) -> ApplicationT: + if application.openapi_config is None: + return application + for route in application.routes: + if not isinstance(route, HTTPRoute): + continue + path_item = (application.openapi_schema.paths or {}).get(route.path_format) + if path_item is None: + continue + for method, (handler, _) in route.route_handler_map.items(): + # A custom handler can own any error schema on this route. + if any( + value != self.exception_response + and (isinstance(key, int) or issubclass(HTTPException, key) or issubclass(key, HTTPException)) + for key, value in handler.resolve_exception_handlers().items() + ): + continue + operation = getattr(path_item, method.lower()) + if operation is None: + continue + self._title_operation(operation) + return application + + @LitestarBootstrapper.use_instrument() class LitestarCorsInstrument(CorsInstrument): def bootstrap_before(self) -> dict[str, typing.Any]: diff --git a/microbootstrap/instruments/http_error_titles_instrument.py b/microbootstrap/instruments/http_error_titles_instrument.py new file mode 100644 index 0000000..f7df8dd --- /dev/null +++ b/microbootstrap/instruments/http_error_titles_instrument.py @@ -0,0 +1,27 @@ +from __future__ import annotations +import http + +from microbootstrap.instruments.base import BaseInstrumentConfig, Instrument + + +class HttpErrorTitlesConfig(BaseInstrumentConfig): + http_error_titles_enabled: bool = False + + +class HttpErrorTitlesInstrument(Instrument[HttpErrorTitlesConfig]): + instrument_name = "HTTP error titles" + ready_condition = "Enable http_error_titles_enabled" + + def is_ready(self) -> bool: + return self.instrument_config.http_error_titles_enabled + + @classmethod + def get_config_type(cls) -> type[HttpErrorTitlesConfig]: + return HttpErrorTitlesConfig + + @staticmethod + def title_for_status(status_code: int) -> str: + try: + return http.HTTPStatus(status_code).phrase + except ValueError: + return "HTTP error" diff --git a/microbootstrap/settings.py b/microbootstrap/settings.py index f7f2400..9fcfcfc 100644 --- a/microbootstrap/settings.py +++ b/microbootstrap/settings.py @@ -12,6 +12,7 @@ FastStreamOpentelemetryConfig, FastStreamPrometheusConfig, HealthChecksConfig, + HttpErrorTitlesConfig, LitestarPrometheusConfig, LoggingConfig, OpentelemetryConfig, @@ -60,6 +61,7 @@ class ServerConfig(pydantic.BaseModel): class LitestarSettings( # type: ignore[misc] BaseServiceSettings, + HttpErrorTitlesConfig, ServerConfig, LoggingConfig, OpentelemetryConfig, @@ -75,6 +77,7 @@ class LitestarSettings( # type: ignore[misc] class FastApiSettings( # type: ignore[misc] BaseServiceSettings, + HttpErrorTitlesConfig, ServerConfig, LoggingConfig, OpentelemetryConfig, diff --git a/tests/bootstrappers/test_http_error_titles.py b/tests/bootstrappers/test_http_error_titles.py new file mode 100644 index 0000000..584278c --- /dev/null +++ b/tests/bootstrappers/test_http_error_titles.py @@ -0,0 +1,303 @@ +import http +import typing + +import fastapi +import litestar +import pytest +from fastapi.testclient import TestClient as FastAPITestClient +from litestar.config.app import AppConfig +from litestar.exceptions import HTTPException as LitestarHTTPException +from litestar.testing import TestClient as LitestarTestClient + +from microbootstrap import FastApiSettings, HttpErrorTitlesConfig, LitestarSettings +from microbootstrap.bootstrappers.fastapi import FastApiBootstrapper +from microbootstrap.bootstrappers.litestar import LitestarBootstrapper +from microbootstrap.config.fastapi import FastApiConfig +from microbootstrap.config.litestar import LitestarConfig + + +if typing.TYPE_CHECKING: + from litestar.types import ExceptionHandlersMap + + +NONSTANDARD_STATUS: typing.Final = 599 + + +@pytest.mark.parametrize("enabled", [False, True]) +@pytest.mark.parametrize("status_code", [http.HTTPStatus.BAD_REQUEST, http.HTTPStatus.UNAUTHORIZED, NONSTANDARD_STATUS]) +def test_fastapi_errors_preserve_fields_headers_and_status(enabled: bool, status_code: int) -> None: + application: typing.Final = FastApiBootstrapper( + FastApiSettings(service_debug=False, http_error_titles_enabled=enabled) + ).bootstrap() + + @application.get("/error") + def reject() -> None: + raise fastapi.HTTPException( + status_code=status_code, detail={"reason": "rejected"}, headers={"X-Reason": "test"} + ) + + with FastAPITestClient(application) as client: + response = client.get("/error") + assert response.status_code == status_code + assert response.headers["x-reason"] == "test" + assert response.json()["detail"] == {"reason": "rejected"} + assert ("title" in response.json()) is enabled + if enabled: + assert response.json()["title"] == ( + "HTTP error" if status_code == NONSTANDARD_STATUS else http.HTTPStatus(status_code).phrase + ) + + +@pytest.mark.parametrize("enabled", [False, True]) +def test_fastapi_validation_schema_matches_real_errors(enabled: bool) -> None: + application: typing.Final = FastApiBootstrapper( + FastApiSettings(service_debug=False, http_error_titles_enabled=enabled) + ).bootstrap() + + @application.get("/items") + def get_items(limit: int) -> int: + return limit + + schema: typing.Final = application.openapi() + assert application.openapi() == schema + error: typing.Final = schema["components"]["schemas"]["HTTPValidationError"] + assert ("title" in error.get("required", [])) is enabled + assert ("title" in error["properties"]) is enabled + with FastAPITestClient(application) as client: + assert client.get("/items?limit=1").json() == 1 + response = client.get("/items?limit=invalid") + assert response.status_code == http.HTTPStatus.UNPROCESSABLE_ENTITY + assert isinstance(response.json()["detail"], list) + assert ("title" in response.json()) is enabled + + +def test_fastapi_empty_schema_and_non_error_exception_keep_native_behavior() -> None: + application: typing.Final = FastApiBootstrapper( + FastApiSettings(service_debug=False, http_error_titles_enabled=True, health_checks_enabled=False) + ).bootstrap() + assert not application.openapi().get("components", {}).get("schemas") + + @application.get("/redirect") + def redirect() -> None: + raise fastapi.HTTPException(status_code=http.HTTPStatus.NOT_MODIFIED, headers={"ETag": "test"}) + + with FastAPITestClient(application) as client: + response = client.get("/redirect") + assert response.status_code == http.HTTPStatus.NOT_MODIFIED + assert response.content == b"" + assert response.headers["etag"] == "test" + + +def test_fastapi_custom_validation_handler_keeps_ownership() -> None: + async def custom_validation(_request: fastapi.Request, _exc: Exception) -> fastapi.responses.JSONResponse: + return fastapi.responses.JSONResponse({"owned": True}, status_code=http.HTTPStatus.BAD_REQUEST) + + application: typing.Final = ( + FastApiBootstrapper(FastApiSettings(service_debug=False, http_error_titles_enabled=True)) + .configure_application( + FastApiConfig(exception_handlers={fastapi.exceptions.RequestValidationError: custom_validation}) + ) + .bootstrap() + ) + + @application.get("/items") + def get_items(limit: int) -> int: + return limit + + assert "title" not in application.openapi()["components"]["schemas"]["HTTPValidationError"].get("required", []) + with FastAPITestClient(application) as client: + assert client.get("/items?limit=1").json() == 1 + assert client.get("/items?limit=invalid").json() == {"owned": True} + + +@pytest.mark.parametrize("enabled", [False, True]) +@pytest.mark.parametrize("media_type", [litestar.MediaType.JSON, litestar.MediaType.HTML]) +def test_litestar_errors_preserve_fields_headers_and_status(enabled: bool, media_type: litestar.MediaType) -> None: + @litestar.get("/error", sync_to_thread=False, media_type=media_type) + def reject() -> None: + raise LitestarHTTPException( + status_code=http.HTTPStatus.UNAUTHORIZED, + detail="rejected", + extra={"reason": "test"}, + headers={"X-Reason": "test"}, + ) + + application: typing.Final = ( + LitestarBootstrapper(LitestarSettings(service_debug=False, http_error_titles_enabled=enabled)) + .configure_application(LitestarConfig(route_handlers=[reject])) + .bootstrap() + ) + with LitestarTestClient(application) as client: + response = client.get("/error") + assert response.status_code == http.HTTPStatus.UNAUTHORIZED + assert response.headers["x-reason"] == "test" + assert response.json()["detail"] == "rejected" + assert response.json()["extra"] == {"reason": "test"} + assert ("title" in response.json()) is enabled + if enabled: + assert response.headers["content-type"].startswith("application/json") + assert response.json()["title"] == "Unauthorized" + + +@pytest.mark.parametrize("enabled", [False, True]) +def test_litestar_validation_schema_matches_real_errors(enabled: bool) -> None: + @litestar.get("/items", sync_to_thread=False) + def get_items(limit: int) -> int: + return limit + + application: typing.Final = ( + LitestarBootstrapper(LitestarSettings(service_debug=False)) + .configure_instrument(HttpErrorTitlesConfig(http_error_titles_enabled=enabled)) + .configure_application(LitestarConfig(route_handlers=[get_items])) + .bootstrap() + ) + schema: typing.Final = application.openapi_schema.to_schema() + error: typing.Final = schema["paths"]["/items"]["get"]["responses"]["400"]["content"]["application/json"]["schema"] + assert ("title" in error["required"]) is enabled + with LitestarTestClient(application) as client: + assert client.get("/items?limit=1").json() == 1 + response = client.get("/items?limit=invalid") + assert response.status_code == http.HTTPStatus.BAD_REQUEST + assert ("title" in response.json()) is enabled + + +@pytest.mark.parametrize("framework", ["fastapi", "litestar"]) +def test_internal_errors_have_safe_titles(framework: str) -> None: + def fail() -> None: + raise RuntimeError("private diagnostic") + + if framework == "fastapi": + application = FastApiBootstrapper( + FastApiSettings(service_debug=False, http_error_titles_enabled=True) + ).bootstrap() + application.get("/error")(fail) + with FastAPITestClient(application, raise_server_exceptions=False) as client: + response = client.get("/error") + else: + handler: typing.Final = litestar.get("/error", sync_to_thread=False)(fail) + litestar_application: typing.Final = ( + LitestarBootstrapper(LitestarSettings(service_debug=False, http_error_titles_enabled=True)) + .configure_application(LitestarConfig(route_handlers=[handler])) + .bootstrap() + ) + with LitestarTestClient(litestar_application, raise_server_exceptions=False) as litestar_client: + response = litestar_client.get("/error") + assert response.status_code == http.HTTPStatus.INTERNAL_SERVER_ERROR + assert response.json()["title"] == "Internal Server Error" + assert "private diagnostic" not in response.text + + +def test_setting_defaults_and_environment(monkeypatch: pytest.MonkeyPatch) -> None: + assert HttpErrorTitlesConfig().http_error_titles_enabled is False + monkeypatch.setenv("HTTP_ERROR_TITLES_ENABLED", "true") + assert FastApiSettings().http_error_titles_enabled is True + assert LitestarSettings().http_error_titles_enabled is True + + +@pytest.mark.parametrize("level", ["application", "controller", "router", "route"]) +@pytest.mark.parametrize("key", [http.HTTPStatus.BAD_REQUEST, litestar.exceptions.ValidationException]) +def test_litestar_custom_validation_handler_owns_payload_and_schema(level: str, key: int | type[Exception]) -> None: + def custom_validation( + _request: litestar.Request[typing.Any, typing.Any, typing.Any], _exc: Exception + ) -> litestar.Response[typing.Any]: + return litestar.Response({"status_code": 400, "detail": "owned", "extra": None}, status_code=400) + + handlers: typing.Final[ExceptionHandlersMap] = {key: custom_validation} + + @litestar.get("/items", sync_to_thread=False, exception_handlers=handlers if level == "route" else None) + def get_items(limit: int) -> int: + return limit + + class ItemsController(litestar.Controller): + path = "/" + exception_handlers = handlers if level == "controller" else None + + @litestar.get("/items", sync_to_thread=False) + def items(self, limit: int) -> int: + return limit + + router: typing.Final = litestar.Router( + path="/", + route_handlers=[ItemsController if level == "controller" else get_items], + exception_handlers=handlers if level == "router" else None, + ) + application: typing.Final = ( + LitestarBootstrapper(LitestarSettings(service_debug=False, http_error_titles_enabled=True)) + .configure_application( + LitestarConfig(route_handlers=[router], exception_handlers=handlers if level == "application" else {}) + ) + .bootstrap() + ) + error: typing.Final = application.openapi_schema.to_schema()["paths"]["/items"]["get"]["responses"]["400"][ + "content" + ]["application/json"]["schema"] + assert "title" not in error["required"] + with LitestarTestClient(application) as client: + response = client.get("/items?limit=invalid") + assert client.get("/items?limit=1").json() == 1 + assert response.status_code == http.HTTPStatus.BAD_REQUEST + assert response.json() == {"status_code": 400, "detail": "owned", "extra": None} + + +@pytest.mark.parametrize("enabled", [False, True]) +def test_litestar_debug_traceback_remains_framework_owned(enabled: bool) -> None: + @litestar.get("/error", sync_to_thread=False) + def fail() -> None: + raise RuntimeError("private diagnostic") + + application: typing.Final = ( + LitestarBootstrapper(LitestarSettings(service_debug=True, http_error_titles_enabled=enabled)) + .configure_application(LitestarConfig(route_handlers=[fail])) + .bootstrap() + ) + with LitestarTestClient(application, raise_server_exceptions=False) as client: + response = client.get("/error") + assert response.status_code == http.HTTPStatus.INTERNAL_SERVER_ERROR + assert "private diagnostic" in response.text + assert response.headers["content-type"].startswith("text/plain") + + +@pytest.mark.parametrize("swagger_enabled", [False, True]) +def test_litestar_non_http_routes_and_combined_error_schemas(swagger_enabled: bool) -> None: + @litestar.get( + "/items", + sync_to_thread=False, + raises=[litestar.exceptions.ValidationException, litestar.exceptions.ClientException, LitestarHTTPException], + ) + def get_items(limit: int) -> int: + return limit + + @litestar.websocket("/ws") + async def websocket_handler(socket: litestar.WebSocket[typing.Any, typing.Any, typing.Any]) -> None: + await socket.accept() + await socket.send_text("ready") + await socket.close() + + def disable_docs(config: AppConfig) -> AppConfig: + config.openapi_config = None + return config + + application: typing.Final = ( + LitestarBootstrapper( + LitestarSettings(service_debug=False, http_error_titles_enabled=True, swagger_enabled=swagger_enabled) + ) + .configure_application( + LitestarConfig( + route_handlers=[get_items, websocket_handler], on_app_init=[] if swagger_enabled else [disable_docs] + ) + ) + .bootstrap() + ) + with LitestarTestClient(application) as client: + with client.websocket_connect("/ws") as websocket: + assert websocket.receive_text() == "ready" + assert client.get("/items?limit=1").json() == 1 + response = client.get("/items?limit=invalid") + assert response.status_code == http.HTTPStatus.BAD_REQUEST + assert response.json()["title"] == "Bad Request" + if swagger_enabled: + schema: typing.Final = application.openapi_schema.to_schema()["paths"]["/items"]["get"]["responses"]["400"][ + "content" + ]["application/json"]["schema"] + assert all("title" in branch["required"] for branch in schema["oneOf"]) + assert "500" in application.openapi_schema.to_schema()["paths"]["/items"]["get"]["responses"] From c1229399d6e13e968d3ad1eecf772695dcf78cea Mon Sep 17 00:00:00 2001 From: Lev Vereshchagin Date: Fri, 23 Oct 2026 00:01:01 +0300 Subject: [PATCH 2/2] Unify API metadata under one bootstrapper setting --- README.md | 47 ++-- microbootstrap/__init__.py | 2 - microbootstrap/bootstrappers/fastapi.py | 164 ++++++------ microbootstrap/bootstrappers/litestar.py | 237 +++++++++--------- microbootstrap/helpers.py | 8 + .../http_error_titles_instrument.py | 27 -- .../instruments/swagger_instrument.py | 1 - microbootstrap/settings.py | 7 +- tests/bootstrappers/test_http_error_titles.py | 62 +++-- .../test_openapi_descriptions.py | 46 ++-- 10 files changed, 296 insertions(+), 305 deletions(-) delete mode 100644 microbootstrap/instruments/http_error_titles_instrument.py diff --git a/README.md b/README.md index f963561..12850d9 100644 --- a/README.md +++ b/README.md @@ -730,44 +730,31 @@ Parameter descriptions: - `health_checks_path` - Path for health check handler. - `health_checks_include_in_schema` - Must be True to include `health_checks_path` (`/health/`) in OpenAPI schema. -### Generated OpenAPI descriptions +### API metadata -Set `openapi_descriptions_enabled=True` in `FastApiSettings`, `LitestarSettings`, -or `SwaggerConfig` to describe framework-generated validation errors and JSON -values automatically. Litestar also fills descriptions for microbootstrap health -fields. The setting is disabled by default and supports the normal environment -prefix, for example `OPENAPI_DESCRIPTIONS_ENABLED=true` without a prefix. +Set `api_metadata_enabled=True` in `FastApiSettings` or `LitestarSettings` to enable +both generated OpenAPI descriptions and HTTP error titles. It defaults to `False` +and can also be enabled with `API_METADATA_ENABLED=true` (with the configured +`ENVIRONMENT_PREFIX`, if any). No separate instrument configuration is required. ```python from microbootstrap import FastApiSettings from microbootstrap.bootstrappers.fastapi import FastApiBootstrapper -app = FastApiBootstrapper(FastApiSettings(openapi_descriptions_enabled=True)).bootstrap() -# Register application routes normally; FastAPI enriches its schema when generated. +app = FastApiBootstrapper(FastApiSettings(api_metadata_enabled=True)).bootstrap() ``` -The bootstrappers preserve existing descriptions and runtime response shapes. -Application-specific descriptions and authentication declarations remain with the -application, using native framework metadata and the existing `security_schemes` -setting. No authentication policy is inferred or enforced by this option. - -### HTTP error titles - -Enable `http_error_titles_enabled=True` in `FastApiSettings`, `LitestarSettings`, -or `HttpErrorTitlesConfig` to add the HTTP status phrase as `title` on default -JSON error responses. The option is disabled by default; the normal settings -environment prefix applies to `HTTP_ERROR_TITLES_ENABLED`. - -The native error fields, status codes and headers are preserved. Default Litestar -errors use JSON even when a route's success media type is different. An unknown -HTTP status receives the title `HTTP error`. Existing application exception -handlers keep precedence and remain responsible for their own payloads and -schemas. Explicit error responses returned by application handlers are also -owned by the application. - -The bootstrappers document `title` as required on their generated validation -schemas when the default handler is active. This is independent of -`openapi_descriptions_enabled`. Debug traceback responses remain framework-owned. +Default HTTP errors receive the HTTP status phrase as `title`, alongside the +framework's existing fields, status and headers. Generated validation schemas +and examples reflect that response. Application, router and route exception +handlers retain ownership of their payloads and schemas. Debug tracebacks retain +the framework's native behavior. Error titles also work with Swagger disabled. +This setting does not document authentication, change successful responses or +certify compliance with an API standard. + +`api_metadata_enabled` replaces the description-only `openapi_descriptions_enabled` +option from 1.8.0. Migrate existing settings and environment variables to the new +name; enabling it also changes default HTTP error bodies as described above. ## Configuration diff --git a/microbootstrap/__init__.py b/microbootstrap/__init__.py index cc53b63..51fe81e 100644 --- a/microbootstrap/__init__.py +++ b/microbootstrap/__init__.py @@ -1,6 +1,5 @@ from microbootstrap.instruments.cors_instrument import CorsConfig from microbootstrap.instruments.health_checks_instrument import HealthChecksConfig -from microbootstrap.instruments.http_error_titles_instrument import HttpErrorTitlesConfig from microbootstrap.instruments.logging_instrument import LoggingConfig from microbootstrap.instruments.openapi_security_schemes import ( OpenApiApiKeySecurityScheme, @@ -51,7 +50,6 @@ "FastStreamSettings", "FastStreamTelemetryMiddlewareProtocol", "HealthChecksConfig", - "HttpErrorTitlesConfig", "InstrumentsSetupperSettings", "LitestarPrometheusConfig", "LitestarSettings", diff --git a/microbootstrap/bootstrappers/fastapi.py b/microbootstrap/bootstrappers/fastapi.py index d5fb519..8bf5f52 100644 --- a/microbootstrap/bootstrappers/fastapi.py +++ b/microbootstrap/bootstrappers/fastapi.py @@ -14,9 +14,9 @@ from microbootstrap.bootstrappers.base import ApplicationBootstrapper from microbootstrap.config.fastapi import FastApiConfig +from microbootstrap.helpers import http_error_title from microbootstrap.instruments.cors_instrument import CorsInstrument from microbootstrap.instruments.health_checks_instrument import HealthChecksInstrument, HealthCheckTypedDict -from microbootstrap.instruments.http_error_titles_instrument import HttpErrorTitlesInstrument from microbootstrap.instruments.logging_instrument import LoggingInstrument from microbootstrap.instruments.openapi_security_schemes import serialize_security_schemes from microbootstrap.instruments.openapi_version_docs import SUPPORTED_HTTP_METHODS @@ -68,6 +68,87 @@ def bootstrap_before(self) -> dict[str, typing.Any]: "lifespan": self._wrapped_lifespan_manager if self.application_config.lifespan else self._lifespan_manager, } + def bootstrap_before_instruments_after_app_created(self, application: ApplicationT) -> ApplicationT: + if self.settings.api_metadata_enabled: + for exception, native, replacement in ( + (HTTPException, fastapi.exception_handlers.http_exception_handler, self.http_exception_response), + ( + RequestValidationError, + fastapi.exception_handlers.request_validation_exception_handler, + self.validation_error_response, + ), + ): + if application.exception_handlers.get(exception) == native: + application.exception_handlers[exception] = replacement + if http.HTTPStatus.INTERNAL_SERVER_ERROR not in application.exception_handlers: + application.exception_handlers.setdefault(Exception, self.internal_error_response) + return application + + async def http_exception_response(self, request: fastapi.Request, exc: HTTPException) -> fastapi.Response: + response = await fastapi.exception_handlers.http_exception_handler(request, exc) + if exc.status_code >= http.HTTPStatus.BAD_REQUEST: + return JSONResponse( + {"detail": exc.detail, "title": http_error_title(exc.status_code)}, + status_code=exc.status_code, + headers=exc.headers, + ) + return response + + async def validation_error_response(self, _request: fastapi.Request, exc: RequestValidationError) -> JSONResponse: + return JSONResponse( + { + "detail": jsonable_encoder(exc.errors()), + "title": http_error_title(http.HTTPStatus.UNPROCESSABLE_ENTITY), + }, + status_code=http.HTTPStatus.UNPROCESSABLE_ENTITY, + ) + + async def internal_error_response(self, _request: fastapi.Request, _exc: Exception) -> JSONResponse: + return JSONResponse( + {"detail": "Internal Server Error", "title": http_error_title(http.HTTPStatus.INTERNAL_SERVER_ERROR)}, + status_code=http.HTTPStatus.INTERNAL_SERVER_ERROR, + ) + + def bootstrap_after(self, application: ApplicationT) -> ApplicationT: + if not self.settings.api_metadata_enabled: + return application + original_openapi: typing.Final = application.openapi + + def titled_openapi() -> dict[str, typing.Any]: + document: typing.Final = original_openapi() + self._describe_generated_schemas(document) + schema = document.get("components", {}).get("schemas", {}).get("HTTPValidationError") + if ( + schema is not None + and application.exception_handlers.get(RequestValidationError) == self.validation_error_response + ): + schema.setdefault("properties", {}).setdefault( + "title", {"type": "string", "description": "Short, human-readable summary of the HTTP error."} + ) + required = schema.setdefault("required", []) + if "title" not in required: + required.append("title") + return document + + application.openapi = titled_openapi # type: ignore[method-assign] # FastAPI's public custom OpenAPI hook. + return application + + @staticmethod + def _describe_generated_schemas(document: dict[str, typing.Any]) -> None: + schemas = document.get("components", {}).get("schemas", {}) + for name in ("HTTPValidationError", "ValidationError"): + if name not in schemas: + continue + definition = schemas[name] + definition.setdefault("description", "Request validation failure reported by FastAPI.") + for field_name, description in _VALIDATION_DESCRIPTIONS.items(): + if field_name in definition.get("properties", {}): + definition["properties"][field_name].setdefault("description", description) + if "JsonValue" in schemas: + schemas["JsonValue"].setdefault( + "description", "JSON value: an object, array, string, number, boolean or null." + ) + FastApiBootstrapper.use_instrument()(SentryInstrument) @@ -85,19 +166,13 @@ def bootstrap_before(self) -> dict[str, typing.Any]: def bootstrap_after(self, application: ApplicationT) -> ApplicationT: if self.instrument_config.swagger_offline_docs: enable_offline_docs(application, static_files_handler=self.instrument_config.service_static_path) - if ( - self.instrument_config.openapi_version_docs is None - and not self.instrument_config.security_schemes - and not self.instrument_config.openapi_descriptions_enabled - ): + if self.instrument_config.openapi_version_docs is None and not self.instrument_config.security_schemes: return application original_openapi: typing.Final = application.openapi def documented_openapi() -> dict[str, typing.Any]: openapi_schema: typing.Final = original_openapi() - if self.instrument_config.openapi_descriptions_enabled: - self._describe_generated_schemas(openapi_schema) if self.instrument_config.security_schemes: self._merge_security_schemes(openapi_schema) if self.instrument_config.openapi_version_docs is not None: @@ -107,22 +182,6 @@ def documented_openapi() -> dict[str, typing.Any]: application.openapi = documented_openapi # type: ignore[method-assign] # FastAPI's public custom OpenAPI hook. return application - @staticmethod - def _describe_generated_schemas(document: dict[str, typing.Any]) -> None: - schemas = document.get("components", {}).get("schemas", {}) - for name in ("HTTPValidationError", "ValidationError"): - if name not in schemas: - continue - definition = schemas[name] - definition.setdefault("description", "Request validation failure reported by FastAPI.") - for field_name, description in _VALIDATION_DESCRIPTIONS.items(): - if field_name in definition.get("properties", {}): - definition["properties"][field_name].setdefault("description", description) - if "JsonValue" in schemas: - schemas["JsonValue"].setdefault( - "description", "JSON value: an object, array, string, number, boolean or null." - ) - def _merge_security_schemes(self, openapi_schema: dict[str, typing.Any]) -> None: expected_schemes: typing.Final = serialize_security_schemes(self.instrument_config.security_schemes) components = openapi_schema.setdefault("components", {}) @@ -150,63 +209,6 @@ def _document_operations(self, openapi_schema: dict[str, typing.Any]) -> None: operation["x-accept-versioning"], operation["description"] = documentation -@FastApiBootstrapper.use_instrument() -class FastApiHttpErrorTitlesInstrument(HttpErrorTitlesInstrument): - async def http_exception_response(self, request: fastapi.Request, exc: HTTPException) -> fastapi.Response: - response = await fastapi.exception_handlers.http_exception_handler(request, exc) - if exc.status_code >= http.HTTPStatus.BAD_REQUEST: - return JSONResponse( - {"detail": exc.detail, "title": self.title_for_status(exc.status_code)}, - status_code=exc.status_code, - headers=exc.headers, - ) - return response - - async def validation_error_response(self, _request: fastapi.Request, exc: RequestValidationError) -> JSONResponse: - return JSONResponse( - { - "detail": jsonable_encoder(exc.errors()), - "title": self.title_for_status(http.HTTPStatus.UNPROCESSABLE_ENTITY), - }, - status_code=http.HTTPStatus.UNPROCESSABLE_ENTITY, - ) - - async def internal_error_response(self, _request: fastapi.Request, _exc: Exception) -> JSONResponse: - return JSONResponse( - {"detail": "Internal Server Error", "title": self.title_for_status(http.HTTPStatus.INTERNAL_SERVER_ERROR)}, - status_code=http.HTTPStatus.INTERNAL_SERVER_ERROR, - ) - - def bootstrap_before(self) -> dict[str, typing.Any]: - return { - "exception_handlers": { - HTTPException: self.http_exception_response, - RequestValidationError: self.validation_error_response, - Exception: self.internal_error_response, - } - } - - def bootstrap_after(self, application: ApplicationT) -> ApplicationT: - if application.exception_handlers.get(RequestValidationError) != self.validation_error_response: - return application - original_openapi: typing.Final = application.openapi - - def titled_openapi() -> dict[str, typing.Any]: - document: typing.Final = original_openapi() - schema = document.get("components", {}).get("schemas", {}).get("HTTPValidationError") - if schema is not None: - schema.setdefault("properties", {}).setdefault( - "title", {"type": "string", "description": "Short, human-readable summary of the HTTP error."} - ) - required = schema.setdefault("required", []) - if "title" not in required: - required.append("title") - return document - - application.openapi = titled_openapi # type: ignore[method-assign] # FastAPI's public custom OpenAPI hook. - return application - - @FastApiBootstrapper.use_instrument() class FastApiCorsInstrument(CorsInstrument): def bootstrap_after(self, application: ApplicationT) -> ApplicationT: diff --git a/microbootstrap/bootstrappers/litestar.py b/microbootstrap/bootstrappers/litestar.py index de3277e..67e9d16 100644 --- a/microbootstrap/bootstrappers/litestar.py +++ b/microbootstrap/bootstrappers/litestar.py @@ -26,13 +26,13 @@ from microbootstrap.bootstrappers.base import ApplicationBootstrapper from microbootstrap.config.litestar import LitestarConfig +from microbootstrap.helpers import http_error_title from microbootstrap.instruments import opentelemetry_instrument from microbootstrap.instruments.cors_instrument import CorsInstrument from microbootstrap.instruments.health_checks_instrument import ( HealthChecksInstrument, HealthCheckTypedDict, ) -from microbootstrap.instruments.http_error_titles_instrument import HttpErrorTitlesInstrument from microbootstrap.instruments.logging_instrument import LoggingInstrument from microbootstrap.instruments.openapi_security_schemes import ( OpenApiApiKeySecurityScheme, @@ -77,6 +77,7 @@ class AcceptVersionedOperation(openapi.spec.Operation): if typing.TYPE_CHECKING: + from litestar.config.app import AppConfig from litestar.contrib.opentelemetry import OpenTelemetryConfig from litestar.types import ASGIApp, Scope from litestar.types.asgi_types import Receive, Send @@ -93,8 +94,120 @@ def bootstrap_before(self: typing_extensions.Self) -> dict[str, typing.Any]: "debug": self.settings.service_debug, "on_shutdown": [self.teardown], "on_startup": [self.console_writer.print_bootstrap_table], + "on_app_init": [self._configure_error_titles] if self.settings.api_metadata_enabled else [], } + def exception_response( + self, request: litestar.Request[typing.Any, typing.Any, typing.Any], exc: Exception + ) -> litestar.Response[typing.Any]: + status_code = exc.status_code if isinstance(exc, HTTPException) else http.HTTPStatus.INTERNAL_SERVER_ERROR + if status_code == http.HTTPStatus.INTERNAL_SERVER_ERROR and request.app.debug: + return create_debug_response(request, exc) + response = create_exception_response(request, exc) + if status_code >= http.HTTPStatus.BAD_REQUEST: + if not isinstance(response.content, dict): + response.content = json.loads(response.content) + response.content.setdefault("title", http_error_title(status_code)) + response.media_type = litestar.MediaType.JSON + return response + + def _configure_error_titles(self, config: AppConfig) -> AppConfig: + config.exception_handlers.setdefault(Exception, self.exception_response) + return config + + @classmethod + def _title_error_schema(cls, schema: openapi.spec.Schema) -> None: + for branch in (*(schema.one_of or []), *(schema.any_of or []), *(schema.all_of or [])): + if isinstance(branch, openapi.spec.Schema): + cls._title_error_schema(branch) + properties = schema.properties or {} + if "status_code" in properties and "detail" in properties: + properties.setdefault( + "title", + openapi.spec.Schema( + type=openapi.spec.OpenAPIType.STRING, + description="Short, human-readable summary of the HTTP error.", + ), + ) + schema.required = list(dict.fromkeys([*(schema.required or []), "title"])) + for example in schema.examples or []: + if isinstance(example, dict) and isinstance(example.get("status_code"), int): + example.setdefault("title", http_error_title(example["status_code"])) + + @classmethod + def _title_operation(cls, operation: openapi.spec.Operation) -> None: + for code, response in (operation.responses or {}).items(): + if str(code).startswith(("4", "5")) and isinstance(response, openapi.spec.OpenAPIResponse): + for media_type in (response.content or {}).values(): + if isinstance(media_type.schema, openapi.spec.Schema): + cls._title_error_schema(media_type.schema) + + def bootstrap_after(self, application: ApplicationT) -> ApplicationT: + if not self.settings.api_metadata_enabled or application.openapi_config is None: + return application + self._describe_generated_schemas(application.openapi_schema) + for route in application.routes: + if not isinstance(route, HTTPRoute): + continue + path_item = (application.openapi_schema.paths or {}).get(route.path_format) + if path_item is None: + continue + for method, (handler, _) in route.route_handler_map.items(): + # A custom handler can own any error schema on this route. + if any( + value != self.exception_response + and (isinstance(key, int) or issubclass(HTTPException, key) or issubclass(key, HTTPException)) + for key, value in handler.resolve_exception_handlers().items() + ): + continue + operation = getattr(path_item, method.lower()) + if operation is None: + continue + self._title_operation(operation) + return application + + @staticmethod + def _describe_properties(schema: openapi.spec.Schema, descriptions: typing.Mapping[str, str]) -> None: + for name, description in descriptions.items(): + field = (schema.properties or {}).get(name) + if field is not None and not field.description: + field.description = description + + @classmethod + def _describe_error_schema(cls, schema: openapi.spec.Schema) -> None: + for branch in (*(schema.one_of or []), *(schema.any_of or []), *(schema.all_of or [])): + if isinstance(branch, openapi.spec.Schema): + cls._describe_error_schema(branch) + if ( + schema.type is None + and schema.one_of + and all( + isinstance(branch, openapi.spec.Schema) and branch.type == openapi.spec.OpenAPIType.OBJECT + for branch in schema.one_of + ) + ): + schema.type = openapi.spec.OpenAPIType.OBJECT + cls._describe_properties(schema, _ERROR_DESCRIPTIONS) + + @classmethod + def _describe_generated_schemas(cls, document: openapi.spec.OpenAPI) -> None: + schemas = document.components.schemas or {} + if "JsonValue" in schemas and not schemas["JsonValue"].description: + schemas["JsonValue"].description = "JSON value: an object, array, string, number, boolean or null." + if "HealthCheckTypedDict" in schemas: + cls._describe_properties(schemas["HealthCheckTypedDict"], _HEALTH_PROPERTIES) + for path_item in (document.paths or {}).values(): + for method in SUPPORTED_HTTP_METHODS: + operation = getattr(path_item, method) + if operation is None: + continue + for code, response in (operation.responses or {}).items(): + if not str(code).startswith(("4", "5")) or not isinstance(response, openapi.spec.OpenAPIResponse): + continue + for media_type in (response.content or {}).values(): + if isinstance(media_type.schema, openapi.spec.Schema): + cls._describe_error_schema(media_type.schema) + @LitestarBootstrapper.use_instrument() class LitestarSentryInstrument(SentryInstrument): @@ -142,62 +255,16 @@ def bootstrap_before(self) -> dict[str, typing.Any]: return bootstrap_result def bootstrap_after(self, application: ApplicationT) -> ApplicationT: - if ( - self.instrument_config.openapi_version_docs is None - and not self.instrument_config.security_schemes - and not self.instrument_config.openapi_descriptions_enabled - ) or (application.openapi_config is None): + if (self.instrument_config.openapi_version_docs is None and not self.instrument_config.security_schemes) or ( + application.openapi_config is None + ): return application - if self.instrument_config.openapi_descriptions_enabled: - self._describe_generated_schemas(application.openapi_schema) if self.instrument_config.security_schemes: self._merge_security_schemes(application.openapi_schema) if self.instrument_config.openapi_version_docs is not None: self._document_operations(application.openapi_schema) return application - @staticmethod - def _describe_properties(schema: openapi.spec.Schema, descriptions: typing.Mapping[str, str]) -> None: - for name, description in descriptions.items(): - field = (schema.properties or {}).get(name) - if field is not None and not field.description: - field.description = description - - @classmethod - def _describe_error_schema(cls, schema: openapi.spec.Schema) -> None: - for branch in (*(schema.one_of or []), *(schema.any_of or []), *(schema.all_of or [])): - if isinstance(branch, openapi.spec.Schema): - cls._describe_error_schema(branch) - if ( - schema.type is None - and schema.one_of - and all( - isinstance(branch, openapi.spec.Schema) and branch.type == openapi.spec.OpenAPIType.OBJECT - for branch in schema.one_of - ) - ): - schema.type = openapi.spec.OpenAPIType.OBJECT - cls._describe_properties(schema, _ERROR_DESCRIPTIONS) - - @classmethod - def _describe_generated_schemas(cls, document: openapi.spec.OpenAPI) -> None: - schemas = document.components.schemas or {} - if "JsonValue" in schemas and not schemas["JsonValue"].description: - schemas["JsonValue"].description = "JSON value: an object, array, string, number, boolean or null." - if "HealthCheckTypedDict" in schemas: - cls._describe_properties(schemas["HealthCheckTypedDict"], _HEALTH_PROPERTIES) - for path_item in (document.paths or {}).values(): - for method in SUPPORTED_HTTP_METHODS: - operation = getattr(path_item, method) - if operation is None: - continue - for code, response in (operation.responses or {}).items(): - if not str(code).startswith(("4", "5")) or not isinstance(response, openapi.spec.OpenAPIResponse): - continue - for media_type in (response.content or {}).values(): - if isinstance(media_type.schema, openapi.spec.Schema): - cls._describe_error_schema(media_type.schema) - def _merge_security_schemes(self, openapi_schema: openapi.spec.OpenAPI) -> None: expected_schemes: typing.Final = serialize_security_schemes(self.instrument_config.security_schemes) security_schemes = openapi_schema.components.security_schemes @@ -284,76 +351,6 @@ def _build_litestar_security_scheme(cls, security_scheme: _OpenApiSecurityScheme return openapi.spec.SecurityScheme(**scheme_data) -@LitestarBootstrapper.use_instrument() -class LitestarHttpErrorTitlesInstrument(HttpErrorTitlesInstrument): - def exception_response( - self, request: litestar.Request[typing.Any, typing.Any, typing.Any], exc: Exception - ) -> litestar.Response[typing.Any]: - status_code = exc.status_code if isinstance(exc, HTTPException) else http.HTTPStatus.INTERNAL_SERVER_ERROR - if status_code == http.HTTPStatus.INTERNAL_SERVER_ERROR and request.app.debug: - return create_debug_response(request, exc) - response = create_exception_response(request, exc) - if status_code >= http.HTTPStatus.BAD_REQUEST: - if not isinstance(response.content, dict): - response.content = json.loads(response.content) - response.content.setdefault("title", self.title_for_status(status_code)) - response.media_type = litestar.MediaType.JSON - return response - - def bootstrap_before(self) -> dict[str, typing.Any]: - return {"exception_handlers": {Exception: self.exception_response}} - - @classmethod - def _title_error_schema(cls, schema: openapi.spec.Schema) -> None: - for branch in (*(schema.one_of or []), *(schema.any_of or []), *(schema.all_of or [])): - if isinstance(branch, openapi.spec.Schema): - cls._title_error_schema(branch) - properties = schema.properties or {} - if "status_code" in properties and "detail" in properties: - properties.setdefault( - "title", - openapi.spec.Schema( - type=openapi.spec.OpenAPIType.STRING, - description="Short, human-readable summary of the HTTP error.", - ), - ) - schema.required = list(dict.fromkeys([*(schema.required or []), "title"])) - for example in schema.examples or []: - if isinstance(example, dict) and isinstance(example.get("status_code"), int): - example.setdefault("title", cls.title_for_status(example["status_code"])) - - @classmethod - def _title_operation(cls, operation: openapi.spec.Operation) -> None: - for code, response in (operation.responses or {}).items(): - if str(code).startswith(("4", "5")) and isinstance(response, openapi.spec.OpenAPIResponse): - for media_type in (response.content or {}).values(): - if isinstance(media_type.schema, openapi.spec.Schema): - cls._title_error_schema(media_type.schema) - - def bootstrap_after(self, application: ApplicationT) -> ApplicationT: - if application.openapi_config is None: - return application - for route in application.routes: - if not isinstance(route, HTTPRoute): - continue - path_item = (application.openapi_schema.paths or {}).get(route.path_format) - if path_item is None: - continue - for method, (handler, _) in route.route_handler_map.items(): - # A custom handler can own any error schema on this route. - if any( - value != self.exception_response - and (isinstance(key, int) or issubclass(HTTPException, key) or issubclass(key, HTTPException)) - for key, value in handler.resolve_exception_handlers().items() - ): - continue - operation = getattr(path_item, method.lower()) - if operation is None: - continue - self._title_operation(operation) - return application - - @LitestarBootstrapper.use_instrument() class LitestarCorsInstrument(CorsInstrument): def bootstrap_before(self) -> dict[str, typing.Any]: diff --git a/microbootstrap/helpers.py b/microbootstrap/helpers.py index 1164b25..aa334ac 100644 --- a/microbootstrap/helpers.py +++ b/microbootstrap/helpers.py @@ -1,4 +1,5 @@ import dataclasses +import http import re import typing @@ -108,3 +109,10 @@ def optimize_exclude_paths( endpoints_to_ignore = set(endpoints_to_ignore) return endpoints_to_ignore + + +def http_error_title(status_code: int) -> str: + try: + return http.HTTPStatus(status_code).phrase + except ValueError: + return "HTTP error" diff --git a/microbootstrap/instruments/http_error_titles_instrument.py b/microbootstrap/instruments/http_error_titles_instrument.py deleted file mode 100644 index f7df8dd..0000000 --- a/microbootstrap/instruments/http_error_titles_instrument.py +++ /dev/null @@ -1,27 +0,0 @@ -from __future__ import annotations -import http - -from microbootstrap.instruments.base import BaseInstrumentConfig, Instrument - - -class HttpErrorTitlesConfig(BaseInstrumentConfig): - http_error_titles_enabled: bool = False - - -class HttpErrorTitlesInstrument(Instrument[HttpErrorTitlesConfig]): - instrument_name = "HTTP error titles" - ready_condition = "Enable http_error_titles_enabled" - - def is_ready(self) -> bool: - return self.instrument_config.http_error_titles_enabled - - @classmethod - def get_config_type(cls) -> type[HttpErrorTitlesConfig]: - return HttpErrorTitlesConfig - - @staticmethod - def title_for_status(status_code: int) -> str: - try: - return http.HTTPStatus(status_code).phrase - except ValueError: - return "HTTP error" diff --git a/microbootstrap/instruments/swagger_instrument.py b/microbootstrap/instruments/swagger_instrument.py index c212b76..00efbda 100644 --- a/microbootstrap/instruments/swagger_instrument.py +++ b/microbootstrap/instruments/swagger_instrument.py @@ -27,7 +27,6 @@ class SwaggerConfig(BaseInstrumentConfig): swagger_extra_params: dict[str, typing.Any] = pydantic.Field(default_factory=dict) security_schemes: typing.Mapping[str, _OpenApiSecurityScheme] = pydantic.Field(default_factory=dict) openapi_version_docs: OpenApiVersionDocsConfig | None = None - openapi_descriptions_enabled: bool = False @pydantic.field_validator("security_schemes") @classmethod diff --git a/microbootstrap/settings.py b/microbootstrap/settings.py index 9fcfcfc..a7cd675 100644 --- a/microbootstrap/settings.py +++ b/microbootstrap/settings.py @@ -12,7 +12,6 @@ FastStreamOpentelemetryConfig, FastStreamPrometheusConfig, HealthChecksConfig, - HttpErrorTitlesConfig, LitestarPrometheusConfig, LoggingConfig, OpentelemetryConfig, @@ -61,7 +60,6 @@ class ServerConfig(pydantic.BaseModel): class LitestarSettings( # type: ignore[misc] BaseServiceSettings, - HttpErrorTitlesConfig, ServerConfig, LoggingConfig, OpentelemetryConfig, @@ -74,10 +72,11 @@ class LitestarSettings( # type: ignore[misc] ): """Settings for a litestar botstrap.""" + api_metadata_enabled: bool = False + class FastApiSettings( # type: ignore[misc] BaseServiceSettings, - HttpErrorTitlesConfig, ServerConfig, LoggingConfig, OpentelemetryConfig, @@ -90,6 +89,8 @@ class FastApiSettings( # type: ignore[misc] ): """Settings for a fastapi botstrap.""" + api_metadata_enabled: bool = False + class FastStreamSettings( # type: ignore[misc] BaseServiceSettings, diff --git a/tests/bootstrappers/test_http_error_titles.py b/tests/bootstrappers/test_http_error_titles.py index 584278c..ae16fa0 100644 --- a/tests/bootstrappers/test_http_error_titles.py +++ b/tests/bootstrappers/test_http_error_titles.py @@ -9,7 +9,7 @@ from litestar.exceptions import HTTPException as LitestarHTTPException from litestar.testing import TestClient as LitestarTestClient -from microbootstrap import FastApiSettings, HttpErrorTitlesConfig, LitestarSettings +from microbootstrap import FastApiSettings, LitestarSettings from microbootstrap.bootstrappers.fastapi import FastApiBootstrapper from microbootstrap.bootstrappers.litestar import LitestarBootstrapper from microbootstrap.config.fastapi import FastApiConfig @@ -27,7 +27,7 @@ @pytest.mark.parametrize("status_code", [http.HTTPStatus.BAD_REQUEST, http.HTTPStatus.UNAUTHORIZED, NONSTANDARD_STATUS]) def test_fastapi_errors_preserve_fields_headers_and_status(enabled: bool, status_code: int) -> None: application: typing.Final = FastApiBootstrapper( - FastApiSettings(service_debug=False, http_error_titles_enabled=enabled) + FastApiSettings(service_debug=False, api_metadata_enabled=enabled) ).bootstrap() @application.get("/error") @@ -51,7 +51,7 @@ def reject() -> None: @pytest.mark.parametrize("enabled", [False, True]) def test_fastapi_validation_schema_matches_real_errors(enabled: bool) -> None: application: typing.Final = FastApiBootstrapper( - FastApiSettings(service_debug=False, http_error_titles_enabled=enabled) + FastApiSettings(service_debug=False, api_metadata_enabled=enabled) ).bootstrap() @application.get("/items") @@ -63,6 +63,7 @@ def get_items(limit: int) -> int: error: typing.Final = schema["components"]["schemas"]["HTTPValidationError"] assert ("title" in error.get("required", [])) is enabled assert ("title" in error["properties"]) is enabled + assert ("description" in error["properties"]["detail"]) is enabled with FastAPITestClient(application) as client: assert client.get("/items?limit=1").json() == 1 response = client.get("/items?limit=invalid") @@ -73,7 +74,7 @@ def get_items(limit: int) -> int: def test_fastapi_empty_schema_and_non_error_exception_keep_native_behavior() -> None: application: typing.Final = FastApiBootstrapper( - FastApiSettings(service_debug=False, http_error_titles_enabled=True, health_checks_enabled=False) + FastApiSettings(service_debug=False, api_metadata_enabled=True, health_checks_enabled=False) ).bootstrap() assert not application.openapi().get("components", {}).get("schemas") @@ -93,7 +94,7 @@ async def custom_validation(_request: fastapi.Request, _exc: Exception) -> fasta return fastapi.responses.JSONResponse({"owned": True}, status_code=http.HTTPStatus.BAD_REQUEST) application: typing.Final = ( - FastApiBootstrapper(FastApiSettings(service_debug=False, http_error_titles_enabled=True)) + FastApiBootstrapper(FastApiSettings(service_debug=False, api_metadata_enabled=True)) .configure_application( FastApiConfig(exception_handlers={fastapi.exceptions.RequestValidationError: custom_validation}) ) @@ -123,7 +124,7 @@ def reject() -> None: ) application: typing.Final = ( - LitestarBootstrapper(LitestarSettings(service_debug=False, http_error_titles_enabled=enabled)) + LitestarBootstrapper(LitestarSettings(service_debug=False, api_metadata_enabled=enabled)) .configure_application(LitestarConfig(route_handlers=[reject])) .bootstrap() ) @@ -146,14 +147,14 @@ def get_items(limit: int) -> int: return limit application: typing.Final = ( - LitestarBootstrapper(LitestarSettings(service_debug=False)) - .configure_instrument(HttpErrorTitlesConfig(http_error_titles_enabled=enabled)) + LitestarBootstrapper(LitestarSettings(service_debug=False, api_metadata_enabled=enabled)) .configure_application(LitestarConfig(route_handlers=[get_items])) .bootstrap() ) schema: typing.Final = application.openapi_schema.to_schema() error: typing.Final = schema["paths"]["/items"]["get"]["responses"]["400"]["content"]["application/json"]["schema"] assert ("title" in error["required"]) is enabled + assert ("description" in error["properties"]["detail"]) is enabled with LitestarTestClient(application) as client: assert client.get("/items?limit=1").json() == 1 response = client.get("/items?limit=invalid") @@ -167,16 +168,14 @@ def fail() -> None: raise RuntimeError("private diagnostic") if framework == "fastapi": - application = FastApiBootstrapper( - FastApiSettings(service_debug=False, http_error_titles_enabled=True) - ).bootstrap() + application = FastApiBootstrapper(FastApiSettings(service_debug=False, api_metadata_enabled=True)).bootstrap() application.get("/error")(fail) with FastAPITestClient(application, raise_server_exceptions=False) as client: response = client.get("/error") else: handler: typing.Final = litestar.get("/error", sync_to_thread=False)(fail) litestar_application: typing.Final = ( - LitestarBootstrapper(LitestarSettings(service_debug=False, http_error_titles_enabled=True)) + LitestarBootstrapper(LitestarSettings(service_debug=False, api_metadata_enabled=True)) .configure_application(LitestarConfig(route_handlers=[handler])) .bootstrap() ) @@ -188,10 +187,11 @@ def fail() -> None: def test_setting_defaults_and_environment(monkeypatch: pytest.MonkeyPatch) -> None: - assert HttpErrorTitlesConfig().http_error_titles_enabled is False - monkeypatch.setenv("HTTP_ERROR_TITLES_ENABLED", "true") - assert FastApiSettings().http_error_titles_enabled is True - assert LitestarSettings().http_error_titles_enabled is True + assert FastApiSettings().api_metadata_enabled is False + assert LitestarSettings().api_metadata_enabled is False + monkeypatch.setenv("API_METADATA_ENABLED", "true") + assert FastApiSettings().api_metadata_enabled is True + assert LitestarSettings().api_metadata_enabled is True @pytest.mark.parametrize("level", ["application", "controller", "router", "route"]) @@ -222,7 +222,7 @@ def items(self, limit: int) -> int: exception_handlers=handlers if level == "router" else None, ) application: typing.Final = ( - LitestarBootstrapper(LitestarSettings(service_debug=False, http_error_titles_enabled=True)) + LitestarBootstrapper(LitestarSettings(service_debug=False, api_metadata_enabled=True)) .configure_application( LitestarConfig(route_handlers=[router], exception_handlers=handlers if level == "application" else {}) ) @@ -246,7 +246,7 @@ def fail() -> None: raise RuntimeError("private diagnostic") application: typing.Final = ( - LitestarBootstrapper(LitestarSettings(service_debug=True, http_error_titles_enabled=enabled)) + LitestarBootstrapper(LitestarSettings(service_debug=True, api_metadata_enabled=enabled)) .configure_application(LitestarConfig(route_handlers=[fail])) .bootstrap() ) @@ -279,7 +279,7 @@ def disable_docs(config: AppConfig) -> AppConfig: application: typing.Final = ( LitestarBootstrapper( - LitestarSettings(service_debug=False, http_error_titles_enabled=True, swagger_enabled=swagger_enabled) + LitestarSettings(service_debug=False, api_metadata_enabled=True, swagger_enabled=swagger_enabled) ) .configure_application( LitestarConfig( @@ -301,3 +301,27 @@ def disable_docs(config: AppConfig) -> AppConfig: ]["application/json"]["schema"] assert all("title" in branch["required"] for branch in schema["oneOf"]) assert "500" in application.openapi_schema.to_schema()["paths"]["/items"]["get"]["responses"] + + +@pytest.mark.parametrize("key", [http.HTTPStatus.INTERNAL_SERVER_ERROR, Exception]) +def test_fastapi_custom_server_error_handler_keeps_ownership(key: int | type[Exception]) -> None: + async def custom_error(_request: fastapi.Request, _exc: Exception) -> fastapi.responses.JSONResponse: + return fastapi.responses.JSONResponse( + {"detail": "owned"}, status_code=http.HTTPStatus.SERVICE_UNAVAILABLE, headers={"X-Owner": "application"} + ) + + application: typing.Final = ( + FastApiBootstrapper(FastApiSettings(service_debug=False, api_metadata_enabled=True)) + .configure_application(FastApiConfig(exception_handlers={key: custom_error})) + .bootstrap() + ) + + @application.get("/error") + def fail() -> None: + raise RuntimeError("private diagnostic") + + with FastAPITestClient(application, raise_server_exceptions=False) as client: + response = client.get("/error") + assert response.status_code == http.HTTPStatus.SERVICE_UNAVAILABLE + assert response.json() == {"detail": "owned"} + assert response.headers["x-owner"] == "application" diff --git a/tests/bootstrappers/test_openapi_descriptions.py b/tests/bootstrappers/test_openapi_descriptions.py index 2e7ac8a..f15f1b1 100644 --- a/tests/bootstrappers/test_openapi_descriptions.py +++ b/tests/bootstrappers/test_openapi_descriptions.py @@ -9,9 +9,9 @@ from litestar.openapi import spec from litestar.testing import TestClient as LitestarTestClient -from microbootstrap import FastApiSettings, LitestarSettings, SwaggerConfig +from microbootstrap import FastApiSettings, LitestarSettings from microbootstrap.bootstrappers.fastapi import FastApiBootstrapper -from microbootstrap.bootstrappers.litestar import LitestarBootstrapper, LitestarSwaggerInstrument +from microbootstrap.bootstrappers.litestar import LitestarBootstrapper from microbootstrap.config.litestar import LitestarConfig from microbootstrap.instruments.health_checks_instrument import HealthCheckTypedDict @@ -22,7 +22,7 @@ class Payload(pydantic.BaseModel): def test_fastapi_descriptions_preserve_validation_and_are_idempotent() -> None: application: typing.Final = FastApiBootstrapper( - FastApiSettings(openapi_descriptions_enabled=True, service_debug=False, health_checks_enabled=False) + FastApiSettings(api_metadata_enabled=True, service_debug=False, health_checks_enabled=False) ).bootstrap() @application.post("/items") @@ -44,19 +44,19 @@ def create_item(payload: Payload) -> Payload: assert client.post("/items", json={"value": {"nested": [1, None]}}).json() == {"value": {"nested": [1, None]}} rejected = client.post("/items", json={}) assert rejected.status_code == http.HTTPStatus.UNPROCESSABLE_ENTITY - assert set(rejected.json()) == {"detail"} - assert "title" not in schemas["HTTPValidationError"]["properties"] + assert set(rejected.json()) == {"detail", "title"} + assert schemas["HTTPValidationError"]["properties"]["title"]["description"] def test_fastapi_empty_schema_is_supported() -> None: application: typing.Final = FastApiBootstrapper( - FastApiSettings(openapi_descriptions_enabled=True, service_debug=False, health_checks_enabled=False) + FastApiSettings(api_metadata_enabled=True, service_debug=False, health_checks_enabled=False) ).bootstrap() assert "security" not in application.openapi() assert not application.openapi().get("components", {}).get("schemas") -def test_litestar_descriptions_preserve_runtime_contracts() -> None: +def test_litestar_metadata_describes_schemas_and_preserves_successful_responses() -> None: @litestar.get("/example-health", sync_to_thread=False) def get_health() -> HealthCheckTypedDict: return {"health_status": True} @@ -66,15 +66,15 @@ def get_item(item_id: int, limit: int = 1) -> int: return item_id * limit application: typing.Final = ( - LitestarBootstrapper(LitestarSettings(openapi_descriptions_enabled=True, service_debug=False)) + LitestarBootstrapper(LitestarSettings(api_metadata_enabled=True, service_debug=False)) .configure_application(LitestarConfig(route_handlers=[get_health, get_item])) .bootstrap() ) document: typing.Final = application.openapi_schema before: typing.Final = copy.deepcopy(document.to_schema()) - LitestarSwaggerInstrument(SwaggerConfig(openapi_descriptions_enabled=True)).bootstrap_after(application) + LitestarBootstrapper(LitestarSettings(api_metadata_enabled=True, service_debug=False)).bootstrap_after(application) after: typing.Final = copy.deepcopy(document.to_schema()) - LitestarSwaggerInstrument(SwaggerConfig(openapi_descriptions_enabled=True)).bootstrap_after(application) + LitestarBootstrapper(LitestarSettings(api_metadata_enabled=True, service_debug=False)).bootstrap_after(application) assert document.to_schema() == after health: typing.Final = after["components"]["schemas"]["HealthCheckTypedDict"] assert health["properties"]["health_status"]["description"].startswith("Static process health") @@ -82,7 +82,7 @@ def get_item(item_id: int, limit: int = 1) -> int: "schema" ] assert error["properties"]["detail"]["description"] - assert "title" not in error["properties"] + assert error["properties"]["title"]["description"] before_error: typing.Final = before["paths"]["/items/{item_id}"]["get"]["responses"]["400"]["content"][ "application/json" ]["schema"] @@ -94,7 +94,7 @@ def get_item(item_id: int, limit: int = 1) -> int: assert client.get(f"/items/{item_id}").json() == item_id rejected = client.get(f"/items/{item_id}?limit=invalid") assert rejected.status_code == http.HTTPStatus.BAD_REQUEST - assert set(rejected.json()) == {"status_code", "detail", "extra"} + assert set(rejected.json()) == {"status_code", "detail", "extra", "title"} def test_composite_error_schemas_and_references_keep_their_shapes() -> None: @@ -137,7 +137,7 @@ def test_composite_error_schemas_and_references_keep_their_shapes() -> None: document: typing.Final = application.openapi_schema document.paths = {"/items": spec.PathItem(get=operation)} document.components.schemas = {"JsonValue": spec.Schema(description="Owned JSON description")} - LitestarSwaggerInstrument(SwaggerConfig(openapi_descriptions_enabled=True)).bootstrap_after(application) + LitestarBootstrapper(LitestarSettings(api_metadata_enabled=True, service_debug=False)).bootstrap_after(application) assert document.components.schemas["JsonValue"].description == "Owned JSON description" assert compound.type == spec.OpenAPIType.OBJECT assert compound.one_of == [branch, spec.Schema(type=spec.OpenAPIType.OBJECT)] @@ -150,7 +150,7 @@ def test_composite_error_schemas_and_references_keep_their_shapes() -> None: def test_fastapi_preserves_owned_descriptions() -> None: application: typing.Final = FastApiBootstrapper( - FastApiSettings(openapi_descriptions_enabled=True, service_debug=False) + FastApiSettings(api_metadata_enabled=True, service_debug=False) ).bootstrap() cached: typing.Final = application.openapi() cached.clear() @@ -178,7 +178,7 @@ def test_fastapi_preserves_owned_descriptions() -> None: @pytest.mark.parametrize("enabled", [False, True]) def test_fastapi_setting_controls_generated_descriptions(enabled: bool) -> None: application: typing.Final = FastApiBootstrapper( - FastApiSettings(openapi_descriptions_enabled=enabled, service_debug=False) + FastApiSettings(api_metadata_enabled=enabled, service_debug=False) ).bootstrap() @application.post("/items") @@ -201,8 +201,7 @@ def get_items(limit: int) -> int: return limit application: typing.Final = ( - LitestarBootstrapper(LitestarSettings(service_debug=False)) - .configure_instrument(SwaggerConfig(openapi_descriptions_enabled=enabled)) + LitestarBootstrapper(LitestarSettings(service_debug=False, api_metadata_enabled=enabled)) .configure_application(LitestarConfig(route_handlers=[get_items])) .bootstrap() ) @@ -217,7 +216,7 @@ def get_items(limit: int) -> int: def test_litestar_missing_openapi_and_empty_schema_are_supported() -> None: - instrument: typing.Final = LitestarSwaggerInstrument(SwaggerConfig(openapi_descriptions_enabled=True)) + instrument: typing.Final = LitestarBootstrapper(LitestarSettings(api_metadata_enabled=True, service_debug=False)) disabled: typing.Final = litestar.Litestar(openapi_config=None) assert instrument.bootstrap_after(disabled) is disabled application: typing.Final = litestar.Litestar(route_handlers=[]) @@ -230,7 +229,10 @@ def test_litestar_missing_openapi_and_empty_schema_are_supported() -> None: def test_environment_can_enable_descriptions(monkeypatch: pytest.MonkeyPatch) -> None: - assert SwaggerConfig().openapi_descriptions_enabled is False - monkeypatch.setenv("OPENAPI_DESCRIPTIONS_ENABLED", "true") - assert FastApiSettings().openapi_descriptions_enabled is True - assert LitestarSettings().openapi_descriptions_enabled is True + assert FastApiSettings().api_metadata_enabled is False + assert LitestarSettings().api_metadata_enabled is False + assert "openapi_descriptions_enabled" not in FastApiSettings.model_fields + assert "openapi_descriptions_enabled" not in LitestarSettings.model_fields + monkeypatch.setenv("API_METADATA_ENABLED", "true") + assert FastApiSettings().api_metadata_enabled is True + assert LitestarSettings().api_metadata_enabled is True