From cad4668bd7d9a38e40c1c6bd6e7cd9d44638d848 Mon Sep 17 00:00:00 2001 From: Lev Vereshchagin Date: Thu, 22 Oct 2026 21:44:59 +0300 Subject: [PATCH 1/3] Add reusable OpenAPI metadata helpers --- README.md | 34 ++++ .../instruments/health_checks_instrument.py | 11 +- microbootstrap/openapi/__init__.py | 1 + microbootstrap/openapi/fastapi.py | 65 ++++++++ microbootstrap/openapi/litestar.py | 102 ++++++++++++ tests/openapi/__init__.py | 0 tests/openapi/test_fastapi.py | 106 +++++++++++++ tests/openapi/test_litestar.py | 149 ++++++++++++++++++ 8 files changed, 465 insertions(+), 3 deletions(-) create mode 100644 microbootstrap/openapi/__init__.py create mode 100644 microbootstrap/openapi/fastapi.py create mode 100644 microbootstrap/openapi/litestar.py create mode 100644 tests/openapi/__init__.py create mode 100644 tests/openapi/test_fastapi.py create mode 100644 tests/openapi/test_litestar.py diff --git a/README.md b/README.md index 3bbc73f..9a3c488 100644 --- a/README.md +++ b/README.md @@ -730,6 +730,40 @@ 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. +### OpenAPI metadata + +Optional helpers fill framework-generated descriptions after routes are registered. +They work with native FastAPI and Litestar applications, including applications +created without a bootstrapper. Install the corresponding framework extra. + +For FastAPI, update the cached document after registering every route: + +```python +from microbootstrap.openapi.fastapi import update_openapi + +update_openapi(app, parameter_descriptions={"item_id": "Identifier of the requested item."}) +``` + +For Litestar, register the helper as an OpenAPI schema creation hook: + +```python +from litestar.openapi.config import OpenAPIConfig +from microbootstrap.openapi.litestar import update_openapi + +openapi_config = OpenAPIConfig( + title="Example API", version="1", on_create_openapi_schema=[update_openapi] +) +``` + +Both helpers describe generated validation errors and JSON values while preserving +existing descriptions and runtime response shapes. Litestar also describes the +health fields owned by microbootstrap. Caller-supplied parameter descriptions apply +to matching names throughout the document. Litestar accepts component and property +descriptions and explicit security overrides keyed by `(lowercase_method, path)`; +FastAPI accepts root security requirements and security schemes, preserving existing +values. Authentication policy and application-specific meanings remain the caller's +responsibility. These declarations document authentication; they do not enforce it. + ## Configuration While settings provide a convenient mechanism, it's not always feasible to store everything within them. 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/openapi/__init__.py b/microbootstrap/openapi/__init__.py new file mode 100644 index 0000000..96545db --- /dev/null +++ b/microbootstrap/openapi/__init__.py @@ -0,0 +1 @@ +"""Optional framework helpers for enriching generated OpenAPI metadata.""" diff --git a/microbootstrap/openapi/fastapi.py b/microbootstrap/openapi/fastapi.py new file mode 100644 index 0000000..793574c --- /dev/null +++ b/microbootstrap/openapi/fastapi.py @@ -0,0 +1,65 @@ +from __future__ import annotations +import typing + + +if typing.TYPE_CHECKING: + from collections.abc import Mapping + + import fastapi + + +_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.", +} +_HTTP_METHODS: typing.Final = ("get", "put", "post", "delete", "options", "head", "patch", "trace") + + +def _update_parameters(document: dict[str, typing.Any], descriptions: Mapping[str, str]) -> None: + for path_item in document.get("paths", {}).values(): + parameter_groups = [path_item.get("parameters", [])] + parameter_groups.extend(path_item.get(method, {}).get("parameters", []) for method in _HTTP_METHODS) + for parameters in parameter_groups: + for parameter in parameters: + description = descriptions.get(parameter.get("name", "")) + if description: + parameter.setdefault("description", description) + + +def update_openapi( + application: fastapi.FastAPI, + *, + security: list[dict[str, list[str]]] | None = None, + security_schemes: Mapping[str, dict[str, typing.Any]] | None = None, + parameter_descriptions: Mapping[str, str] | None = None, +) -> None: + """Enrich the cached schema after all routes are registered, preserving existing metadata. + + Authentication declarations and parameter meanings come from the caller; + this helper neither infers authorization nor changes request handling. + Parameter descriptions apply to matching names throughout the document. + """ + document: typing.Final = application.openapi() + schemas: typing.Final = document.setdefault("components", {}).setdefault("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.") + if security is not None: + document.setdefault("security", security) + if security_schemes: + definitions = document["components"].setdefault("securitySchemes", {}) + for name, scheme in security_schemes.items(): + definitions.setdefault(name, scheme) + if parameter_descriptions: + _update_parameters(document, parameter_descriptions) diff --git a/microbootstrap/openapi/litestar.py b/microbootstrap/openapi/litestar.py new file mode 100644 index 0000000..d5fc70c --- /dev/null +++ b/microbootstrap/openapi/litestar.py @@ -0,0 +1,102 @@ +from __future__ import annotations +import typing + +import pydantic +from litestar.openapi import spec + +from microbootstrap.instruments.health_checks_instrument import HealthCheckTypedDict + + +if typing.TYPE_CHECKING: + from collections.abc import Iterable, Mapping + + +_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() +} +_HTTP_METHODS: typing.Final = ("get", "put", "post", "delete", "options", "head", "patch", "trace") + + +def _update_properties(schema: spec.Schema, descriptions: Mapping[str, str]) -> None: + for name, description in descriptions.items(): + field = (schema.properties or {}).get(name) + if isinstance(field, spec.Schema) and not field.description: + field.description = description + + +def _update_error(schema: spec.Schema) -> None: + for branch in (*(schema.one_of or []), *(schema.any_of or []), *(schema.all_of or [])): + if isinstance(branch, spec.Schema): + _update_error(branch) + if ( + schema.type is None + and schema.one_of + and all(isinstance(branch, spec.Schema) and branch.type == spec.OpenAPIType.OBJECT for branch in schema.one_of) + ): + schema.type = spec.OpenAPIType.OBJECT + _update_properties(schema, _ERROR_DESCRIPTIONS) + + +def _update_components( + document: spec.OpenAPI, + descriptions: Mapping[str, str | None], + properties: Mapping[str, Mapping[str, str]], +) -> None: + for name, schema in (document.components.schemas or {}).items(): + description = descriptions.get(name) + if description and not schema.description: + schema.description = description + default_properties = _HEALTH_PROPERTIES if name == "HealthCheckTypedDict" else {} + _update_properties(schema, default_properties | dict(properties.get(name, {}))) + + +def _update_parameters(parameters: Iterable[spec.Parameter | spec.Reference], descriptions: Mapping[str, str]) -> None: + for parameter in parameters: + if isinstance(parameter, spec.Parameter) and not parameter.description: + parameter.description = descriptions.get(parameter.name) + + +def _update_operation(operation: spec.Operation, parameter_descriptions: Mapping[str, str]) -> None: + _update_parameters(operation.parameters or [], parameter_descriptions) + for code, response in (operation.responses or {}).items(): + if not str(code).startswith(("4", "5")) or not isinstance(response, spec.OpenAPIResponse): + continue + for media_type in (response.content or {}).values(): + if isinstance(media_type.schema, spec.Schema): + _update_error(media_type.schema) + + +def update_openapi( + document: spec.OpenAPI, + *, + schema_descriptions: Mapping[str, str | None] | None = None, + property_descriptions: Mapping[str, Mapping[str, str]] | None = None, + parameter_descriptions: Mapping[str, str] | None = None, + operation_security: Mapping[tuple[str, str], list[spec.SecurityRequirement]] | None = None, +) -> None: + """Describe generated errors and fill caller-supplied metadata without changing handlers. + + Security overrides use (lowercase HTTP method, exact OpenAPI path) keys. + No authentication scheme or domain-specific field meaning is inferred. + """ + _update_components( + document, + {"JsonValue": "JSON value: an object, array, string, number, boolean or null."} + | dict(schema_descriptions or {}), + property_descriptions or {}, + ) + for path, path_item in (document.paths or {}).items(): + _update_parameters(path_item.parameters or [], parameter_descriptions or {}) + for method in _HTTP_METHODS: + operation = getattr(path_item, method) + if operation is None: + continue + if operation_security is not None and (method, path) in operation_security: + operation.security = operation_security[method, path] + _update_operation(operation, parameter_descriptions or {}) diff --git a/tests/openapi/__init__.py b/tests/openapi/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/openapi/test_fastapi.py b/tests/openapi/test_fastapi.py new file mode 100644 index 0000000..2be6370 --- /dev/null +++ b/tests/openapi/test_fastapi.py @@ -0,0 +1,106 @@ +import copy +import http +import typing + +import fastapi +import pydantic +from fastapi.testclient import TestClient + +from microbootstrap.openapi.fastapi import update_openapi + + +class Payload(pydantic.BaseModel): + value: pydantic.JsonValue + + +def test_generated_metadata_preserves_validation_and_is_idempotent() -> None: + application: typing.Final = fastapi.FastAPI() + + @application.post("/items") + def create_item(payload: Payload) -> Payload: + return payload + + before: typing.Final = copy.deepcopy(application.openapi()) + update_openapi(application) + after: typing.Final = copy.deepcopy(application.openapi()) + update_openapi(application) + 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 TestClient(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_no_routes_or_schema_components_are_required() -> None: + application: typing.Final = fastapi.FastAPI() + update_openapi(application, security=[], security_schemes={}, parameter_descriptions={}) + assert application.openapi()["security"] == [] + assert application.openapi()["components"]["schemas"] == {} + + +def test_caller_metadata_preserves_existing_descriptions_and_security() -> None: + application: typing.Final = fastapi.FastAPI() + document: typing.Final = application.openapi() + document.update( + { + "security": [{"existing": []}], + "components": { + "schemas": { + "HTTPValidationError": { + "description": "Owned model", + "properties": {"detail": {"description": "Owned field"}}, + }, + "JsonValue": {"description": "Owned JSON description"}, + }, + "securitySchemes": {"existing": {"type": "http", "scheme": "bearer"}}, + }, + "paths": { + "/items": { + "parameters": [{"name": "shared", "in": "query"}], + "get": { + "parameters": [ + {"name": "item_id", "in": "query"}, + {"name": "owned", "in": "query", "description": "Owned parameter"}, + {"$ref": "#/components/parameters/External"}, + {"name": "unmapped", "in": "query"}, + ] + }, + "post": {"parameters": [{"name": "item_id", "in": "query"}]}, + } + }, + } + ) + update_openapi( + application, + security=[], + security_schemes={ + "existing": {"type": "apiKey", "in": "header", "name": "X-Key"}, + "new": {"type": "http", "scheme": "bearer"}, + }, + parameter_descriptions={"shared": "Shared value", "item_id": "Item identifier", "owned": "Replacement"}, + ) + assert document["security"] == [{"existing": []}] + assert document["components"]["securitySchemes"]["existing"]["type"] == "http" + assert document["components"]["securitySchemes"]["new"]["scheme"] == "bearer" + 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" + path_item: typing.Final = document["paths"]["/items"] + assert path_item["parameters"][0]["description"] == "Shared value" + assert path_item["get"]["parameters"][0]["description"] == "Item identifier" + assert path_item["post"]["parameters"][0]["description"] == "Item identifier" + assert path_item["get"]["parameters"][1]["description"] == "Owned parameter" + assert "description" not in path_item["get"]["parameters"][2] + assert "description" not in path_item["get"]["parameters"][3] diff --git a/tests/openapi/test_litestar.py b/tests/openapi/test_litestar.py new file mode 100644 index 0000000..a91ad34 --- /dev/null +++ b/tests/openapi/test_litestar.py @@ -0,0 +1,149 @@ +import copy +import http +import typing + +import litestar +from litestar.openapi import spec +from litestar.testing import TestClient + +from microbootstrap.instruments.health_checks_instrument import HealthCheckTypedDict +from microbootstrap.openapi.litestar import update_openapi + + +def _document( + *, + components: spec.Components | None = None, + paths: dict[str, spec.PathItem] | None = None, +) -> spec.OpenAPI: + return spec.OpenAPI( + info=spec.Info(title="Example API", version="1"), + servers=[], + components=components or spec.Components(), + paths=paths, + ) + + +def test_real_errors_and_health_schemas_preserve_runtime_contracts() -> None: + @litestar.get("/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 = litestar.Litestar(route_handlers=[get_health, get_item]) + document: typing.Final = application.openapi_schema + before: typing.Final = copy.deepcopy(document.to_schema()) + update_openapi(document) + after: typing.Final = copy.deepcopy(document.to_schema()) + update_openapi(document) + 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 TestClient(application) as client: + 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_empty_document_and_components_are_supported() -> None: + update_openapi(_document()) + update_openapi(_document(components=spec.Components())) + + +def test_component_and_operation_overrides_are_explicit_and_preserve_owned_metadata() -> None: + field: typing.Final = spec.Schema(type=spec.OpenAPIType.STRING) + owned_field: typing.Final = spec.Schema(description="Owned field") + schema: typing.Final = spec.Schema( + properties={"value": field, "owned": owned_field, "reference": spec.Reference(ref="#/components/schemas/Other")} + ) + owned: typing.Final = spec.Schema(description="Owned schema") + parameter: typing.Final = spec.Parameter(name="item_id", param_in="query") + shared: typing.Final = spec.Parameter(name="shared", param_in="query") + owned_parameter: typing.Final = spec.Parameter(name="owned", param_in="query", description="Owned parameter") + unmapped: typing.Final = spec.Parameter(name="unmapped", param_in="query") + protected: typing.Final = spec.Operation(security=[{"existing": []}]) + selected: typing.Final = spec.Operation( + parameters=[parameter, owned_parameter, unmapped, spec.Reference(ref="#/components/parameters/External")] + ) + document: typing.Final = _document( + components=spec.Components(schemas={"Custom": schema, "Owned": owned, "JsonValue": spec.Schema()}), + paths={"/items": spec.PathItem(get=protected, post=selected, parameters=[shared])}, + ) + update_openapi( + document, + schema_descriptions={"Custom": "Caller schema", "Owned": "Replacement", "Absent": None}, + property_descriptions={"Custom": {"value": "Caller value", "owned": "Replacement", "missing": "Ignored"}}, + parameter_descriptions={"item_id": "Item identifier", "shared": "Shared value", "owned": "Replacement"}, + operation_security={("post", "/items"): [], ("delete", "/absent"): []}, + ) + assert schema.description == "Caller schema" + assert field.description == "Caller value" + assert owned.description == "Owned schema" + assert owned_field.description == "Owned field" + assert parameter.description == "Item identifier" + assert shared.description == "Shared value" + assert owned_parameter.description == "Owned parameter" + assert unmapped.description is None + assert selected.security == [] + assert protected.security == [{"existing": []}] + + +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"), + } + ) + document: typing.Final = _document(paths={"/items": spec.PathItem(get=operation)}) + update_openapi(document) + 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 From af1680fa9738069d602ec7d6845591e3cefdda46 Mon Sep 17 00:00:00 2001 From: Lev Vereshchagin Date: Thu, 22 Oct 2026 21:48:02 +0300 Subject: [PATCH 2/3] Correct Litestar usage example and verify health response --- README.md | 8 +++----- tests/openapi/test_litestar.py | 1 + 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 9a3c488..daa1d3e 100644 --- a/README.md +++ b/README.md @@ -744,15 +744,13 @@ from microbootstrap.openapi.fastapi import update_openapi update_openapi(app, parameter_descriptions={"item_id": "Identifier of the requested item."}) ``` -For Litestar, register the helper as an OpenAPI schema creation hook: +For Litestar, update the generated document after constructing the application +with OpenAPI enabled: ```python -from litestar.openapi.config import OpenAPIConfig from microbootstrap.openapi.litestar import update_openapi -openapi_config = OpenAPIConfig( - title="Example API", version="1", on_create_openapi_schema=[update_openapi] -) +update_openapi(app.openapi_schema) ``` Both helpers describe generated validation errors and JSON values while preserving diff --git a/tests/openapi/test_litestar.py b/tests/openapi/test_litestar.py index a91ad34..dc4d8b8 100644 --- a/tests/openapi/test_litestar.py +++ b/tests/openapi/test_litestar.py @@ -53,6 +53,7 @@ def get_item(item_id: int, limit: int = 1) -> int: assert set(error["properties"]) == set(before_error["properties"]) item_id: typing.Final = 42 with TestClient(application) as client: + assert client.get("/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 From ce1959a7eacfd767b435649a7fde72ea777d7063 Mon Sep 17 00:00:00 2001 From: Lev Vereshchagin Date: Thu, 22 Oct 2026 21:56:52 +0300 Subject: [PATCH 3/3] Integrate generated descriptions into bootstrappers behind a setting --- README.md | 39 ++- microbootstrap/bootstrappers/fastapi.py | 34 ++- microbootstrap/bootstrappers/litestar.py | 64 ++++- .../instruments/swagger_instrument.py | 1 + microbootstrap/openapi/__init__.py | 1 - microbootstrap/openapi/fastapi.py | 65 ----- microbootstrap/openapi/litestar.py | 102 -------- .../test_openapi_descriptions.py | 236 ++++++++++++++++++ tests/openapi/__init__.py | 0 tests/openapi/test_fastapi.py | 106 -------- tests/openapi/test_litestar.py | 150 ----------- 11 files changed, 345 insertions(+), 453 deletions(-) delete mode 100644 microbootstrap/openapi/__init__.py delete mode 100644 microbootstrap/openapi/fastapi.py delete mode 100644 microbootstrap/openapi/litestar.py create mode 100644 tests/bootstrappers/test_openapi_descriptions.py delete mode 100644 tests/openapi/__init__.py delete mode 100644 tests/openapi/test_fastapi.py delete mode 100644 tests/openapi/test_litestar.py diff --git a/README.md b/README.md index daa1d3e..ad80697 100644 --- a/README.md +++ b/README.md @@ -730,37 +730,26 @@ 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. -### OpenAPI metadata +### Generated OpenAPI descriptions -Optional helpers fill framework-generated descriptions after routes are registered. -They work with native FastAPI and Litestar applications, including applications -created without a bootstrapper. Install the corresponding framework extra. - -For FastAPI, update the cached document after registering every route: - -```python -from microbootstrap.openapi.fastapi import update_openapi - -update_openapi(app, parameter_descriptions={"item_id": "Identifier of the requested item."}) -``` - -For Litestar, update the generated document after constructing the application -with OpenAPI enabled: +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.openapi.litestar import update_openapi +from microbootstrap import FastApiSettings +from microbootstrap.bootstrappers.fastapi import FastApiBootstrapper -update_openapi(app.openapi_schema) +app = FastApiBootstrapper(FastApiSettings(openapi_descriptions_enabled=True)).bootstrap() +# Register application routes normally; FastAPI enriches its schema when generated. ``` -Both helpers describe generated validation errors and JSON values while preserving -existing descriptions and runtime response shapes. Litestar also describes the -health fields owned by microbootstrap. Caller-supplied parameter descriptions apply -to matching names throughout the document. Litestar accepts component and property -descriptions and explicit security overrides keyed by `(lowercase_method, path)`; -FastAPI accepts root security requirements and security schemes, preserving existing -values. Authentication policy and application-specific meanings remain the caller's -responsibility. These declarations document authentication; they do not enforce it. +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 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/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/microbootstrap/openapi/__init__.py b/microbootstrap/openapi/__init__.py deleted file mode 100644 index 96545db..0000000 --- a/microbootstrap/openapi/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Optional framework helpers for enriching generated OpenAPI metadata.""" diff --git a/microbootstrap/openapi/fastapi.py b/microbootstrap/openapi/fastapi.py deleted file mode 100644 index 793574c..0000000 --- a/microbootstrap/openapi/fastapi.py +++ /dev/null @@ -1,65 +0,0 @@ -from __future__ import annotations -import typing - - -if typing.TYPE_CHECKING: - from collections.abc import Mapping - - import fastapi - - -_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.", -} -_HTTP_METHODS: typing.Final = ("get", "put", "post", "delete", "options", "head", "patch", "trace") - - -def _update_parameters(document: dict[str, typing.Any], descriptions: Mapping[str, str]) -> None: - for path_item in document.get("paths", {}).values(): - parameter_groups = [path_item.get("parameters", [])] - parameter_groups.extend(path_item.get(method, {}).get("parameters", []) for method in _HTTP_METHODS) - for parameters in parameter_groups: - for parameter in parameters: - description = descriptions.get(parameter.get("name", "")) - if description: - parameter.setdefault("description", description) - - -def update_openapi( - application: fastapi.FastAPI, - *, - security: list[dict[str, list[str]]] | None = None, - security_schemes: Mapping[str, dict[str, typing.Any]] | None = None, - parameter_descriptions: Mapping[str, str] | None = None, -) -> None: - """Enrich the cached schema after all routes are registered, preserving existing metadata. - - Authentication declarations and parameter meanings come from the caller; - this helper neither infers authorization nor changes request handling. - Parameter descriptions apply to matching names throughout the document. - """ - document: typing.Final = application.openapi() - schemas: typing.Final = document.setdefault("components", {}).setdefault("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.") - if security is not None: - document.setdefault("security", security) - if security_schemes: - definitions = document["components"].setdefault("securitySchemes", {}) - for name, scheme in security_schemes.items(): - definitions.setdefault(name, scheme) - if parameter_descriptions: - _update_parameters(document, parameter_descriptions) diff --git a/microbootstrap/openapi/litestar.py b/microbootstrap/openapi/litestar.py deleted file mode 100644 index d5fc70c..0000000 --- a/microbootstrap/openapi/litestar.py +++ /dev/null @@ -1,102 +0,0 @@ -from __future__ import annotations -import typing - -import pydantic -from litestar.openapi import spec - -from microbootstrap.instruments.health_checks_instrument import HealthCheckTypedDict - - -if typing.TYPE_CHECKING: - from collections.abc import Iterable, Mapping - - -_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() -} -_HTTP_METHODS: typing.Final = ("get", "put", "post", "delete", "options", "head", "patch", "trace") - - -def _update_properties(schema: spec.Schema, descriptions: Mapping[str, str]) -> None: - for name, description in descriptions.items(): - field = (schema.properties or {}).get(name) - if isinstance(field, spec.Schema) and not field.description: - field.description = description - - -def _update_error(schema: spec.Schema) -> None: - for branch in (*(schema.one_of or []), *(schema.any_of or []), *(schema.all_of or [])): - if isinstance(branch, spec.Schema): - _update_error(branch) - if ( - schema.type is None - and schema.one_of - and all(isinstance(branch, spec.Schema) and branch.type == spec.OpenAPIType.OBJECT for branch in schema.one_of) - ): - schema.type = spec.OpenAPIType.OBJECT - _update_properties(schema, _ERROR_DESCRIPTIONS) - - -def _update_components( - document: spec.OpenAPI, - descriptions: Mapping[str, str | None], - properties: Mapping[str, Mapping[str, str]], -) -> None: - for name, schema in (document.components.schemas or {}).items(): - description = descriptions.get(name) - if description and not schema.description: - schema.description = description - default_properties = _HEALTH_PROPERTIES if name == "HealthCheckTypedDict" else {} - _update_properties(schema, default_properties | dict(properties.get(name, {}))) - - -def _update_parameters(parameters: Iterable[spec.Parameter | spec.Reference], descriptions: Mapping[str, str]) -> None: - for parameter in parameters: - if isinstance(parameter, spec.Parameter) and not parameter.description: - parameter.description = descriptions.get(parameter.name) - - -def _update_operation(operation: spec.Operation, parameter_descriptions: Mapping[str, str]) -> None: - _update_parameters(operation.parameters or [], parameter_descriptions) - for code, response in (operation.responses or {}).items(): - if not str(code).startswith(("4", "5")) or not isinstance(response, spec.OpenAPIResponse): - continue - for media_type in (response.content or {}).values(): - if isinstance(media_type.schema, spec.Schema): - _update_error(media_type.schema) - - -def update_openapi( - document: spec.OpenAPI, - *, - schema_descriptions: Mapping[str, str | None] | None = None, - property_descriptions: Mapping[str, Mapping[str, str]] | None = None, - parameter_descriptions: Mapping[str, str] | None = None, - operation_security: Mapping[tuple[str, str], list[spec.SecurityRequirement]] | None = None, -) -> None: - """Describe generated errors and fill caller-supplied metadata without changing handlers. - - Security overrides use (lowercase HTTP method, exact OpenAPI path) keys. - No authentication scheme or domain-specific field meaning is inferred. - """ - _update_components( - document, - {"JsonValue": "JSON value: an object, array, string, number, boolean or null."} - | dict(schema_descriptions or {}), - property_descriptions or {}, - ) - for path, path_item in (document.paths or {}).items(): - _update_parameters(path_item.parameters or [], parameter_descriptions or {}) - for method in _HTTP_METHODS: - operation = getattr(path_item, method) - if operation is None: - continue - if operation_security is not None and (method, path) in operation_security: - operation.security = operation_security[method, path] - _update_operation(operation, parameter_descriptions or {}) 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 diff --git a/tests/openapi/__init__.py b/tests/openapi/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/tests/openapi/test_fastapi.py b/tests/openapi/test_fastapi.py deleted file mode 100644 index 2be6370..0000000 --- a/tests/openapi/test_fastapi.py +++ /dev/null @@ -1,106 +0,0 @@ -import copy -import http -import typing - -import fastapi -import pydantic -from fastapi.testclient import TestClient - -from microbootstrap.openapi.fastapi import update_openapi - - -class Payload(pydantic.BaseModel): - value: pydantic.JsonValue - - -def test_generated_metadata_preserves_validation_and_is_idempotent() -> None: - application: typing.Final = fastapi.FastAPI() - - @application.post("/items") - def create_item(payload: Payload) -> Payload: - return payload - - before: typing.Final = copy.deepcopy(application.openapi()) - update_openapi(application) - after: typing.Final = copy.deepcopy(application.openapi()) - update_openapi(application) - 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 TestClient(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_no_routes_or_schema_components_are_required() -> None: - application: typing.Final = fastapi.FastAPI() - update_openapi(application, security=[], security_schemes={}, parameter_descriptions={}) - assert application.openapi()["security"] == [] - assert application.openapi()["components"]["schemas"] == {} - - -def test_caller_metadata_preserves_existing_descriptions_and_security() -> None: - application: typing.Final = fastapi.FastAPI() - document: typing.Final = application.openapi() - document.update( - { - "security": [{"existing": []}], - "components": { - "schemas": { - "HTTPValidationError": { - "description": "Owned model", - "properties": {"detail": {"description": "Owned field"}}, - }, - "JsonValue": {"description": "Owned JSON description"}, - }, - "securitySchemes": {"existing": {"type": "http", "scheme": "bearer"}}, - }, - "paths": { - "/items": { - "parameters": [{"name": "shared", "in": "query"}], - "get": { - "parameters": [ - {"name": "item_id", "in": "query"}, - {"name": "owned", "in": "query", "description": "Owned parameter"}, - {"$ref": "#/components/parameters/External"}, - {"name": "unmapped", "in": "query"}, - ] - }, - "post": {"parameters": [{"name": "item_id", "in": "query"}]}, - } - }, - } - ) - update_openapi( - application, - security=[], - security_schemes={ - "existing": {"type": "apiKey", "in": "header", "name": "X-Key"}, - "new": {"type": "http", "scheme": "bearer"}, - }, - parameter_descriptions={"shared": "Shared value", "item_id": "Item identifier", "owned": "Replacement"}, - ) - assert document["security"] == [{"existing": []}] - assert document["components"]["securitySchemes"]["existing"]["type"] == "http" - assert document["components"]["securitySchemes"]["new"]["scheme"] == "bearer" - 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" - path_item: typing.Final = document["paths"]["/items"] - assert path_item["parameters"][0]["description"] == "Shared value" - assert path_item["get"]["parameters"][0]["description"] == "Item identifier" - assert path_item["post"]["parameters"][0]["description"] == "Item identifier" - assert path_item["get"]["parameters"][1]["description"] == "Owned parameter" - assert "description" not in path_item["get"]["parameters"][2] - assert "description" not in path_item["get"]["parameters"][3] diff --git a/tests/openapi/test_litestar.py b/tests/openapi/test_litestar.py deleted file mode 100644 index dc4d8b8..0000000 --- a/tests/openapi/test_litestar.py +++ /dev/null @@ -1,150 +0,0 @@ -import copy -import http -import typing - -import litestar -from litestar.openapi import spec -from litestar.testing import TestClient - -from microbootstrap.instruments.health_checks_instrument import HealthCheckTypedDict -from microbootstrap.openapi.litestar import update_openapi - - -def _document( - *, - components: spec.Components | None = None, - paths: dict[str, spec.PathItem] | None = None, -) -> spec.OpenAPI: - return spec.OpenAPI( - info=spec.Info(title="Example API", version="1"), - servers=[], - components=components or spec.Components(), - paths=paths, - ) - - -def test_real_errors_and_health_schemas_preserve_runtime_contracts() -> None: - @litestar.get("/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 = litestar.Litestar(route_handlers=[get_health, get_item]) - document: typing.Final = application.openapi_schema - before: typing.Final = copy.deepcopy(document.to_schema()) - update_openapi(document) - after: typing.Final = copy.deepcopy(document.to_schema()) - update_openapi(document) - 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 TestClient(application) as client: - assert client.get("/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_empty_document_and_components_are_supported() -> None: - update_openapi(_document()) - update_openapi(_document(components=spec.Components())) - - -def test_component_and_operation_overrides_are_explicit_and_preserve_owned_metadata() -> None: - field: typing.Final = spec.Schema(type=spec.OpenAPIType.STRING) - owned_field: typing.Final = spec.Schema(description="Owned field") - schema: typing.Final = spec.Schema( - properties={"value": field, "owned": owned_field, "reference": spec.Reference(ref="#/components/schemas/Other")} - ) - owned: typing.Final = spec.Schema(description="Owned schema") - parameter: typing.Final = spec.Parameter(name="item_id", param_in="query") - shared: typing.Final = spec.Parameter(name="shared", param_in="query") - owned_parameter: typing.Final = spec.Parameter(name="owned", param_in="query", description="Owned parameter") - unmapped: typing.Final = spec.Parameter(name="unmapped", param_in="query") - protected: typing.Final = spec.Operation(security=[{"existing": []}]) - selected: typing.Final = spec.Operation( - parameters=[parameter, owned_parameter, unmapped, spec.Reference(ref="#/components/parameters/External")] - ) - document: typing.Final = _document( - components=spec.Components(schemas={"Custom": schema, "Owned": owned, "JsonValue": spec.Schema()}), - paths={"/items": spec.PathItem(get=protected, post=selected, parameters=[shared])}, - ) - update_openapi( - document, - schema_descriptions={"Custom": "Caller schema", "Owned": "Replacement", "Absent": None}, - property_descriptions={"Custom": {"value": "Caller value", "owned": "Replacement", "missing": "Ignored"}}, - parameter_descriptions={"item_id": "Item identifier", "shared": "Shared value", "owned": "Replacement"}, - operation_security={("post", "/items"): [], ("delete", "/absent"): []}, - ) - assert schema.description == "Caller schema" - assert field.description == "Caller value" - assert owned.description == "Owned schema" - assert owned_field.description == "Owned field" - assert parameter.description == "Item identifier" - assert shared.description == "Shared value" - assert owned_parameter.description == "Owned parameter" - assert unmapped.description is None - assert selected.security == [] - assert protected.security == [{"existing": []}] - - -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"), - } - ) - document: typing.Final = _document(paths={"/items": spec.PathItem(get=operation)}) - update_openapi(document) - 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