diff --git a/README.md b/README.md index 3bbc73f..ad80697 100644 --- a/README.md +++ b/README.md @@ -730,6 +730,27 @@ 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 + +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. + +```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. +``` + +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. + ## Configuration While settings provide a convenient mechanism, it's not always feasible to store everything within them. diff --git a/microbootstrap/bootstrappers/fastapi.py b/microbootstrap/bootstrappers/fastapi.py index 48988ea..fd54101 100644 --- a/microbootstrap/bootstrappers/fastapi.py +++ b/microbootstrap/bootstrappers/fastapi.py @@ -23,6 +23,16 @@ from microbootstrap.settings import FastApiSettings +_VALIDATION_DESCRIPTIONS: typing.Final = { + "detail": "Validation errors describing the rejected request fields.", + "loc": "Field names and array indices locating the invalid input.", + "msg": "Human-readable validation error message.", + "type": "Machine-readable validation error category.", + "input": "Input value associated with the validation failure.", + "ctx": "Additional context explaining the validation error.", +} + + ApplicationT = typing.TypeVar("ApplicationT", bound=fastapi.FastAPI) @@ -69,13 +79,19 @@ 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: + if ( + self.instrument_config.openapi_version_docs is None + and not self.instrument_config.security_schemes + and not self.instrument_config.openapi_descriptions_enabled + ): 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: @@ -85,6 +101,22 @@ 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", {}) diff --git a/microbootstrap/bootstrappers/litestar.py b/microbootstrap/bootstrappers/litestar.py index ab129e3..df77906 100644 --- a/microbootstrap/bootstrappers/litestar.py +++ b/microbootstrap/bootstrappers/litestar.py @@ -3,6 +3,7 @@ import typing import litestar +import pydantic import typing_extensions from litestar import openapi from litestar.config.cors import CORSConfig as LitestarCorsConfig @@ -47,6 +48,17 @@ from microbootstrap.settings import LitestarSettings +_ERROR_DESCRIPTIONS: typing.Final = { + "status_code": "HTTP status code of the failed request.", + "detail": "Human-readable explanation of the request failure.", + "extra": "Additional validation or error details supplied by Litestar.", +} +_HEALTH_PROPERTIES: typing.Final[dict[str, str]] = { + name: field["description"] + for name, field in pydantic.TypeAdapter(HealthCheckTypedDict).json_schema()["properties"].items() +} + + ApplicationT = typing.TypeVar("ApplicationT", bound=litestar.Litestar) @@ -124,16 +136,62 @@ 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) or ( - application.openapi_config is None - ): + 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): 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 diff --git a/microbootstrap/instruments/health_checks_instrument.py b/microbootstrap/instruments/health_checks_instrument.py index 9b0ee3f..982fa3d 100644 --- a/microbootstrap/instruments/health_checks_instrument.py +++ b/microbootstrap/instruments/health_checks_instrument.py @@ -1,14 +1,19 @@ from __future__ import annotations +import typing +import pydantic import typing_extensions from microbootstrap.instruments.base import BaseInstrumentConfig, Instrument class HealthCheckTypedDict(typing_extensions.TypedDict, total=False): - service_version: str | None - service_name: str | None - health_status: bool + service_version: typing.Annotated[str | None, pydantic.Field(description="Version of the running service.")] + service_name: typing.Annotated[str | None, pydantic.Field(description="Name of the running service.")] + health_status: typing.Annotated[ + bool, + pydantic.Field(description="Static process health status; external dependencies are not checked."), + ] class HealthChecksConfig(BaseInstrumentConfig): diff --git a/microbootstrap/instruments/swagger_instrument.py b/microbootstrap/instruments/swagger_instrument.py index 00efbda..c212b76 100644 --- a/microbootstrap/instruments/swagger_instrument.py +++ b/microbootstrap/instruments/swagger_instrument.py @@ -27,6 +27,7 @@ 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/tests/bootstrappers/test_openapi_descriptions.py b/tests/bootstrappers/test_openapi_descriptions.py new file mode 100644 index 0000000..2e7ac8a --- /dev/null +++ b/tests/bootstrappers/test_openapi_descriptions.py @@ -0,0 +1,236 @@ +import copy +import http +import typing + +import litestar +import pydantic +import pytest +from fastapi.testclient import TestClient as FastAPITestClient +from litestar.openapi import spec +from litestar.testing import TestClient as LitestarTestClient + +from microbootstrap import FastApiSettings, LitestarSettings, SwaggerConfig +from microbootstrap.bootstrappers.fastapi import FastApiBootstrapper +from microbootstrap.bootstrappers.litestar import LitestarBootstrapper, LitestarSwaggerInstrument +from microbootstrap.config.litestar import LitestarConfig +from microbootstrap.instruments.health_checks_instrument import HealthCheckTypedDict + + +class Payload(pydantic.BaseModel): + value: pydantic.JsonValue + + +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) + ).bootstrap() + + @application.post("/items") + def create_item(payload: Payload) -> Payload: + return payload + + before: typing.Final = copy.deepcopy(application.openapi()) + after: typing.Final = copy.deepcopy(application.openapi()) + assert application.openapi() == after + assert "security" not in after + schemas: typing.Final = after["components"]["schemas"] + assert schemas["JsonValue"]["description"].startswith("JSON value:") + assert schemas["HTTPValidationError"]["properties"]["detail"]["description"] + assert schemas["ValidationError"]["properties"]["loc"]["description"] + for name in ("HTTPValidationError", "ValidationError"): + assert schemas[name].get("required") == before["components"]["schemas"][name].get("required") + assert set(schemas[name]["properties"]) == set(before["components"]["schemas"][name]["properties"]) + with FastAPITestClient(application) as client: + 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"] + + +def test_fastapi_empty_schema_is_supported() -> None: + application: typing.Final = FastApiBootstrapper( + FastApiSettings(openapi_descriptions_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: + @litestar.get("/example-health", sync_to_thread=False) + def get_health() -> HealthCheckTypedDict: + return {"health_status": True} + + @litestar.get("/items/{item_id:int}", sync_to_thread=False) + 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)) + .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) + after: typing.Final = copy.deepcopy(document.to_schema()) + LitestarSwaggerInstrument(SwaggerConfig(openapi_descriptions_enabled=True)).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") + error: typing.Final = after["paths"]["/items/{item_id}"]["get"]["responses"]["400"]["content"]["application/json"][ + "schema" + ] + assert error["properties"]["detail"]["description"] + assert "title" not in error["properties"] + before_error: typing.Final = before["paths"]["/items/{item_id}"]["get"]["responses"]["400"]["content"][ + "application/json" + ]["schema"] + assert error["required"] == before_error["required"] + assert set(error["properties"]) == set(before_error["properties"]) + item_id: typing.Final = 42 + with LitestarTestClient(application) as client: + assert client.get("/example-health").json() == {"health_status": True} + 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"} + + +def test_composite_error_schemas_and_references_keep_their_shapes() -> None: + detail: typing.Final = spec.Schema(type=spec.OpenAPIType.STRING) + owned: typing.Final = spec.Schema(description="Owned detail") + branch: typing.Final = spec.Schema(type=spec.OpenAPIType.OBJECT, properties={"detail": detail}) + compound: typing.Final = spec.Schema( + one_of=[branch, spec.Schema(type=spec.OpenAPIType.OBJECT)], + any_of=[spec.Schema(properties={"detail": owned})], + all_of=[spec.Reference(ref="#/components/schemas/Other")], + ) + mixed: typing.Final = spec.Schema( + one_of=[spec.Reference(ref="#/components/schemas/Other"), spec.Schema(type=spec.OpenAPIType.STRING)] + ) + success: typing.Final = spec.Schema(properties={"detail": spec.Schema()}) + operation: typing.Final = spec.Operation( + responses={ + "200": spec.OpenAPIResponse( + description="OK", content={"application/json": spec.OpenAPIMediaType(schema=success)} + ), + "400": spec.OpenAPIResponse( + description="Bad request", content={"application/json": spec.OpenAPIMediaType(schema=compound)} + ), + "401": spec.OpenAPIResponse( + description="Unauthorized", content={"application/json": spec.OpenAPIMediaType(schema=mixed)} + ), + "402": spec.Reference(ref="#/components/responses/Payment"), + "403": spec.OpenAPIResponse( + description="Forbidden", + content={ + "application/json": spec.OpenAPIMediaType( + schema=spec.Reference(ref="#/components/schemas/Forbidden") + ) + }, + ), + "500": spec.OpenAPIResponse(description="Internal error"), + } + ) + application: typing.Final = litestar.Litestar(route_handlers=[]) + 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) + 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)] + assert detail.description + assert owned.description == "Owned detail" + assert mixed.type is None + assert success.properties is not None + assert success.properties["detail"].description is None + + +def test_fastapi_preserves_owned_descriptions() -> None: + application: typing.Final = FastApiBootstrapper( + FastApiSettings(openapi_descriptions_enabled=True, service_debug=False) + ).bootstrap() + cached: typing.Final = application.openapi() + cached.clear() + cached.update( + { + "components": { + "schemas": { + "HTTPValidationError": { + "description": "Owned model", + "properties": {"detail": {"description": "Owned field"}}, + }, + "JsonValue": {"description": "Owned JSON description"}, + } + }, + } + ) + document: typing.Final = application.openapi() + assert document["components"]["schemas"]["HTTPValidationError"]["description"] == "Owned model" + assert ( + document["components"]["schemas"]["HTTPValidationError"]["properties"]["detail"]["description"] == "Owned field" + ) + assert document["components"]["schemas"]["JsonValue"]["description"] == "Owned JSON description" + + +@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) + ).bootstrap() + + @application.post("/items") + def create_item(payload: Payload) -> Payload: + return payload + + document: typing.Final = application.openapi() + assert ("description" in document["components"]["schemas"]["HTTPValidationError"]) is enabled + assert "security" not in document + with FastAPITestClient(application) as client: + assert client.post("/items", json={"value": 1}).json() == {"value": 1} + application.openapi_schema = None + assert application.openapi() == document + + +@pytest.mark.parametrize("enabled", [False, True]) +def test_litestar_setting_controls_generated_descriptions(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(SwaggerConfig(openapi_descriptions_enabled=enabled)) + .configure_application(LitestarConfig(route_handlers=[get_items])) + .bootstrap() + ) + document: typing.Final = application.openapi_schema.to_schema() + error: typing.Final = document["paths"]["/items"]["get"]["responses"]["400"]["content"]["application/json"][ + "schema" + ] + assert ("description" in error["properties"]["detail"]) is enabled + assert "security" not in document + with LitestarTestClient(application) as client: + assert client.get("/items?limit=1").json() == 1 + + +def test_litestar_missing_openapi_and_empty_schema_are_supported() -> None: + instrument: typing.Final = LitestarSwaggerInstrument(SwaggerConfig(openapi_descriptions_enabled=True)) + disabled: typing.Final = litestar.Litestar(openapi_config=None) + assert instrument.bootstrap_after(disabled) is disabled + application: typing.Final = litestar.Litestar(route_handlers=[]) + application.openapi_schema.components.schemas = {"JsonValue": spec.Schema()} + application.openapi_schema.paths = None + instrument.bootstrap_after(application) + assert application.openapi_schema.components.schemas["JsonValue"].description + application.openapi_schema.components.schemas = {} + instrument.bootstrap_after(application) + + +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