From 59919226b6ee0af834015a45c81b167e28c9a9a2 Mon Sep 17 00:00:00 2001 From: Lev Vereshchagin Date: Mon, 2 Nov 2026 14:50:01 +0300 Subject: [PATCH 1/5] Preserve Pydantic descriptions in Litestar schemas --- README.md | 5 + microbootstrap/bootstrappers/litestar.py | 15 ++ .../litestar_pydantic_descriptions.py | 77 ++++++++ .../test_litestar_pydantic_descriptions.py | 187 ++++++++++++++++++ 4 files changed, 284 insertions(+) create mode 100644 microbootstrap/instruments/litestar_pydantic_descriptions.py create mode 100644 tests/bootstrappers/test_litestar_pydantic_descriptions.py diff --git a/README.md b/README.md index 12850d9..40f07f6 100644 --- a/README.md +++ b/README.md @@ -737,6 +737,11 @@ 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. +For Litestar, this also preserves Pydantic model docstrings and `Field(description=...)` +metadata on `TypedDict` properties during schema generation, including nested and +referenced models. Explicit descriptions and application-provided schema plugins +take precedence. Validation, serialization and field constraints remain unchanged. + ```python from microbootstrap import FastApiSettings from microbootstrap.bootstrappers.fastapi import FastApiBootstrapper diff --git a/microbootstrap/bootstrappers/litestar.py b/microbootstrap/bootstrappers/litestar.py index 67e9d16..cd08ff3 100644 --- a/microbootstrap/bootstrappers/litestar.py +++ b/microbootstrap/bootstrappers/litestar.py @@ -17,6 +17,7 @@ from litestar.exceptions.responses import create_debug_response, create_exception_response from litestar.middleware import ASGIMiddleware from litestar.openapi.plugins import SwaggerRenderPlugin +from litestar.plugins.pydantic import PydanticPlugin from litestar.routes import HTTPRoute from litestar.types.asgi_types import ASGIApp, Scope from litestar_offline_docs import generate_static_files_config @@ -33,6 +34,10 @@ HealthChecksInstrument, HealthCheckTypedDict, ) +from microbootstrap.instruments.litestar_pydantic_descriptions import ( + PydanticDescriptionsSchemaPlugin, + PydanticTypedDictDescriptionsPlugin, +) from microbootstrap.instruments.logging_instrument import LoggingInstrument from microbootstrap.instruments.openapi_security_schemes import ( OpenApiApiKeySecurityScheme, @@ -90,11 +95,21 @@ class LitestarBootstrapper( application_type = litestar.Litestar def bootstrap_before(self: typing_extensions.Self) -> dict[str, typing.Any]: + prefer_alias: typing.Final = next( + (plugin.prefer_alias for plugin in self.application_config.plugins if isinstance(plugin, PydanticPlugin)), + False, + ) return { "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 [], + "plugins": [ + PydanticDescriptionsSchemaPlugin(prefer_alias=prefer_alias), + PydanticTypedDictDescriptionsPlugin(), + ] + if self.settings.api_metadata_enabled + else [], } def exception_response( diff --git a/microbootstrap/instruments/litestar_pydantic_descriptions.py b/microbootstrap/instruments/litestar_pydantic_descriptions.py new file mode 100644 index 0000000..486880d --- /dev/null +++ b/microbootstrap/instruments/litestar_pydantic_descriptions.py @@ -0,0 +1,77 @@ +from __future__ import annotations +import dataclasses +import inspect +import typing + +import pydantic +from litestar.params import KwargDefinition, ParameterKwarg +from litestar.plugins import OpenAPISchemaPlugin +from litestar.plugins.pydantic import PydanticSchemaPlugin +from litestar.typing import FieldDefinition +from pydantic.fields import FieldInfo + + +if typing.TYPE_CHECKING: + from litestar._openapi.schema_generation import SchemaCreator + from litestar.openapi import spec + + +def _fetch_field_description(field_definition: FieldDefinition) -> str | None: + keyword_definition: typing.Final = field_definition.kwarg_definition + if isinstance(keyword_definition, KwargDefinition) and keyword_definition.description: + return keyword_definition.description + return next( + ( + metadata.description + for metadata in reversed(field_definition.metadata) + if isinstance(metadata, FieldInfo) and metadata.description + ), + None, + ) + + +@typing.final +class PydanticDescriptionsSchemaPlugin(PydanticSchemaPlugin): + def to_openapi_schema(self, field_definition: FieldDefinition, schema_creator: SchemaCreator) -> spec.Schema: + generated_schema: typing.Final = super().to_openapi_schema(field_definition, schema_creator) + model_type: typing.Final = field_definition.annotation + if isinstance(model_type, type) and issubclass(model_type, pydantic.BaseModel): + model_description: typing.Final = model_type.__dict__.get("__doc__") + if not generated_schema.description: + generated_schema.description = _fetch_field_description(field_definition) or ( + inspect.cleandoc(model_description) if isinstance(model_description, str) else None + ) + return generated_schema + + +def _describe_typed_dict_field(field_name: str, annotation: object) -> FieldDefinition: + original_field: typing.Final = FieldDefinition.from_kwarg(annotation, field_name) + original_keyword: typing.Final = original_field.kwarg_definition + description: typing.Final = _fetch_field_description(original_field) + if not description or (original_keyword is not None and not isinstance(original_keyword, KwargDefinition)): + return original_field + return dataclasses.replace( + original_field, + kwarg_definition=dataclasses.replace( + original_keyword if isinstance(original_keyword, KwargDefinition) else ParameterKwarg(), + description=description, + ), + ) + + +@typing.final +class PydanticTypedDictDescriptionsPlugin(OpenAPISchemaPlugin): + def is_plugin_supported_field(self, field_definition: FieldDefinition) -> bool: + return field_definition.is_typeddict_type + + def to_openapi_schema(self, field_definition: FieldDefinition, schema_creator: SchemaCreator) -> spec.Schema: + return schema_creator.create_component_schema( + field_definition, + required=sorted(field_definition.type_.__required_keys__), + property_fields={ + field_name: _describe_typed_dict_field(field_name, annotation) + for field_name, annotation in field_definition.get_type_hints( + include_extras=True, resolve_generics=True + ).items() + }, + ) diff --git a/tests/bootstrappers/test_litestar_pydantic_descriptions.py b/tests/bootstrappers/test_litestar_pydantic_descriptions.py new file mode 100644 index 0000000..f6ac1af --- /dev/null +++ b/tests/bootstrappers/test_litestar_pydantic_descriptions.py @@ -0,0 +1,187 @@ +from __future__ import annotations +import copy +import http +import typing + +import litestar +import pydantic +import pytest +import typing_extensions +from litestar.params import Parameter +from litestar.plugins.pydantic import PydanticPlugin, PydanticSchemaPlugin +from litestar.testing import TestClient + +from microbootstrap import LitestarSettings +from microbootstrap.bootstrappers.litestar import LitestarBootstrapper +from microbootstrap.config.litestar import LitestarConfig + + +if typing.TYPE_CHECKING: + from litestar._openapi.schema_generation import SchemaCreator + from litestar.handlers import HTTPRouteHandler + from litestar.openapi import spec + from litestar.plugins import PluginProtocol + from litestar.typing import FieldDefinition + + +@typing.final +class DocumentedRecord(pydantic.BaseModel): + """One record returned by the catalog.""" + + identifier: str = pydantic.Field(description="Record identifier.") + + +@typing.final +class RecordTree(pydantic.BaseModel): + """A tree of catalog records.""" + + children: list[RecordTree] = pydantic.Field(default_factory=list, description="Nested records.") + + +IntervalBoundaries = typing_extensions.TypedDict( + "IntervalBoundaries", + { + "from": typing.Annotated[int | None, pydantic.Field(description="Lower interval bound.")], + "to": typing.Annotated[ + int | None, + pydantic.Field(description="Pydantic interval description."), + Parameter(description="Explicit upper interval bound."), + ], + "label": typing_extensions.NotRequired[typing.Annotated[str, pydantic.Field(description="Interval label.")]], + "unlabelled": typing_extensions.NotRequired[str], + }, +) + + +@typing.final +class RecordPayload(pydantic.BaseModel): + """Catalog records and their selection interval.""" + + records: list[DocumentedRecord] = pydantic.Field(description="Selected records.") + selected: DocumentedRecord | None = pydantic.Field(default=None, description="Selected record, when available.") + interval: IntervalBoundaries = pydantic.Field(description="Selection boundaries.") + record_tree: RecordTree = pydantic.Field(description="Record hierarchy.") + source_url: pydantic.AnyHttpUrl = pydantic.Field(description="Catalog source.") + limit: int = pydantic.Field(ge=1, le=3, description="Maximum selected records.") + + +@typing.final +class AliasedRecord(pydantic.BaseModel): + """An aliased catalog record.""" + + record_id: str = pydantic.Field(alias="recordId", description="Public record identifier.") + + +@litestar.post("/records", sync_to_thread=False) +def create_records(data: RecordPayload) -> RecordPayload: + return data + + +def _build_application( + route_handler: HTTPRouteHandler, + *, + metadata_enabled: bool = True, + plugins: list[PluginProtocol] | None = None, +) -> litestar.Litestar: + return ( + LitestarBootstrapper( + LitestarSettings(api_metadata_enabled=metadata_enabled, service_debug=False, health_checks_enabled=False) + ) + .configure_application(LitestarConfig(route_handlers=[route_handler], plugins=plugins or [])) + .bootstrap() + ) + + +@pytest.mark.parametrize("metadata_enabled", [False, True]) +def test_litestar_describes_nested_models_and_typed_dict_fields(metadata_enabled: bool) -> None: + application: typing.Final = _build_application(create_records, metadata_enabled=metadata_enabled) + document: typing.Final = application.openapi_schema.to_schema() + component_schemas: typing.Final = document["components"]["schemas"] + assert bool(component_schemas["RecordPayload"].get("description")) is metadata_enabled + assert bool(component_schemas["DocumentedRecord"].get("description")) is metadata_enabled + assert component_schemas["RecordTree"]["description"] == "Record hierarchy." + interval_schema: typing.Final = component_schemas["IntervalBoundaries"] + assert bool(interval_schema["properties"]["from"].get("description")) is metadata_enabled + assert interval_schema["properties"]["to"]["description"] == "Explicit upper interval bound." + assert bool(interval_schema["properties"]["label"].get("description")) is metadata_enabled + assert "description" not in interval_schema["properties"]["unlabelled"] + assert interval_schema["required"] == ["from", "to"] + assert component_schemas["RecordTree"]["properties"]["children"]["items"] == { + "$ref": "#/components/schemas/RecordTree" + } + assert document == application.openapi_schema.to_schema() + + +def _strip_descriptions(schema_value: object) -> object: + if isinstance(schema_value, dict): + return {key: _strip_descriptions(value) for key, value in schema_value.items() if key != "description"} + if isinstance(schema_value, list): + return [_strip_descriptions(value) for value in schema_value] + return schema_value + + +@pytest.mark.parametrize("metadata_enabled", [False, True]) +def test_litestar_descriptions_preserve_constraints_and_wire_values(metadata_enabled: bool) -> None: + request_payload: typing.Final = { + "records": [{"identifier": "catalog-entry"}], + "selected": {"identifier": "catalog-entry"}, + "interval": {"from": None, "to": 2}, + "record_tree": {"children": [{"children": []}]}, + "source_url": "https://example.org/", + "limit": 2, + } + baseline: typing.Final = _build_application(create_records, metadata_enabled=False) + described: typing.Final = _build_application(create_records) + baseline_document: typing.Final = baseline.openapi_schema.to_schema() + described_document: typing.Final = described.openapi_schema.to_schema() + for schema_name in ("RecordPayload", "DocumentedRecord", "RecordTree", "IntervalBoundaries"): + assert _strip_descriptions(baseline_document["components"]["schemas"][schema_name]) == _strip_descriptions( + described_document["components"]["schemas"][schema_name] + ) + with TestClient(described if metadata_enabled else baseline) as test_client: + accepted: typing.Final = test_client.post("/records", json=request_payload) + invalid_payload: typing.Final = copy.deepcopy(request_payload) + invalid_payload["limit"] = 0 + rejected: typing.Final = test_client.post("/records", json=invalid_payload) + assert accepted.status_code == http.HTTPStatus.CREATED + assert accepted.json() == request_payload + assert rejected.status_code == http.HTTPStatus.BAD_REQUEST + + +def test_litestar_preserves_explicit_model_description() -> None: + @litestar.get("/record", sync_to_thread=False) + def fetch_record() -> typing.Annotated[DocumentedRecord, Parameter(description="Explicit record contract.")]: + return DocumentedRecord(identifier="catalog-entry") + + application: typing.Final = _build_application(fetch_record) + document: typing.Final = application.openapi_schema.to_schema() + assert document["components"]["schemas"]["DocumentedRecord"]["description"] == "Explicit record contract." + with TestClient(application) as test_client: + assert test_client.get("/record").json() == {"identifier": "catalog-entry"} + + +def test_litestar_preserves_custom_schema_plugin() -> None: + @typing.final + class CustomRecordPlugin(PydanticSchemaPlugin): + def to_openapi_schema(self, field_definition: FieldDefinition, schema_creator: SchemaCreator) -> spec.Schema: + generated_schema: typing.Final = super().to_openapi_schema(field_definition, schema_creator) + generated_schema.description = "Custom record contract." + return generated_schema + + application: typing.Final = _build_application(create_records, plugins=[CustomRecordPlugin()]) + document: typing.Final = application.openapi_schema.to_schema() + assert document["components"]["schemas"]["DocumentedRecord"]["description"] == "Custom record contract." + + +def test_litestar_preserves_pydantic_plugin_aliases() -> None: + @litestar.get("/record", sync_to_thread=False) + def fetch_record() -> AliasedRecord: + return AliasedRecord(recordId="catalog-entry") + + application: typing.Final = _build_application(fetch_record, plugins=[PydanticPlugin(prefer_alias=True)]) + document: typing.Final = application.openapi_schema.to_schema() + record_schema: typing.Final = document["components"]["schemas"]["AliasedRecord"] + assert set(record_schema["properties"]) == {"recordId"} + assert record_schema["description"] == "An aliased catalog record." + with TestClient(application) as test_client: + assert test_client.get("/record").json() == {"recordId": "catalog-entry"} From 6d460204f9ff24eeaa41553299c35a11e68a995d Mon Sep 17 00:00:00 2001 From: Lev Vereshchagin Date: Mon, 2 Nov 2026 14:53:49 +0300 Subject: [PATCH 2/5] Document the static Litestar health check --- microbootstrap/bootstrappers/litestar.py | 9 ++++++++- tests/instruments/test_health_checks.py | 16 +++++++++++++++- 2 files changed, 23 insertions(+), 2 deletions(-) diff --git a/microbootstrap/bootstrappers/litestar.py b/microbootstrap/bootstrappers/litestar.py index cd08ff3..8a03e69 100644 --- a/microbootstrap/bootstrappers/litestar.py +++ b/microbootstrap/bootstrappers/litestar.py @@ -487,7 +487,14 @@ def get_config_type(cls) -> type[LitestarPrometheusConfig]: @LitestarBootstrapper.use_instrument() class LitestarHealthChecksInstrument(HealthChecksInstrument): def build_litestar_health_check_router(self) -> litestar.Router: - @litestar.get(media_type=litestar.MediaType.JSON) + @litestar.get( + media_type=litestar.MediaType.JSON, + security=[], + description=( + "Return the service name, version and static health status for this running process. " + "This endpoint does not check external dependencies." + ), + ) async def health_check_handler() -> HealthCheckTypedDict: return self.render_health_check_data() diff --git a/tests/instruments/test_health_checks.py b/tests/instruments/test_health_checks.py index b87c111..50f0a27 100644 --- a/tests/instruments/test_health_checks.py +++ b/tests/instruments/test_health_checks.py @@ -42,7 +42,9 @@ def test_health_checks_teardown( def test_litestar_health_checks_bootstrap() -> None: test_health_checks_path: typing.Final = "/test-path/" - heatlh_checks_config: typing.Final = HealthChecksConfig(health_checks_path=test_health_checks_path) + heatlh_checks_config: typing.Final = HealthChecksConfig( + health_checks_path=test_health_checks_path, health_checks_include_in_schema=True + ) health_checks_instrument: typing.Final = LitestarHealthChecksInstrument(heatlh_checks_config) health_checks_instrument.bootstrap() @@ -53,6 +55,18 @@ def test_litestar_health_checks_bootstrap() -> None: with LitestarTestClient(app=litestar_application) as async_client: response = async_client.get(heatlh_checks_config.health_checks_path) assert response.status_code == status_codes.HTTP_200_OK + assert response.json() == { + "service_name": heatlh_checks_config.service_name, + "service_version": heatlh_checks_config.service_version, + "health_status": True, + } + + document: typing.Final = litestar_application.openapi_schema.to_schema() + health_operation: typing.Final = document["paths"][test_health_checks_path.rstrip("/")]["get"] + assert "static health status" in health_operation["description"] + assert "does not check external dependencies" in health_operation["description"] + assert not document.get("security") + assert not health_operation.get("security") def test_fastapi_health_checks_bootstrap() -> None: From 9499d1f6f339c799f003f440bf4a2863dfe74d56 Mon Sep 17 00:00:00 2001 From: Lev Vereshchagin Date: Mon, 2 Nov 2026 15:15:23 +0300 Subject: [PATCH 3/5] Generate native schemas for Pydantic root values --- README.md | 2 + microbootstrap/bootstrappers/litestar.py | 6 +- .../litestar_pydantic_descriptions.py | 69 +++--- .../litestar_typed_dict_descriptions.py | 47 ++++ .../test_litestar_root_models.py | 210 ++++++++++++++++++ 5 files changed, 288 insertions(+), 46 deletions(-) create mode 100644 microbootstrap/instruments/litestar_typed_dict_descriptions.py create mode 100644 tests/bootstrappers/test_litestar_root_models.py diff --git a/README.md b/README.md index 40f07f6..b86ae6b 100644 --- a/README.md +++ b/README.md @@ -741,6 +741,8 @@ For Litestar, this also preserves Pydantic model docstrings and `Field(descripti metadata on `TypedDict` properties during schema generation, including nested and referenced models. Explicit descriptions and application-provided schema plugins take precedence. Validation, serialization and field constraints remain unchanged. +Pydantic `RootModel` schemas describe the root value rather than a `root` property, +including roots nested inside other models and recursive root collections. ```python from microbootstrap import FastApiSettings diff --git a/microbootstrap/bootstrappers/litestar.py b/microbootstrap/bootstrappers/litestar.py index 8a03e69..8353b57 100644 --- a/microbootstrap/bootstrappers/litestar.py +++ b/microbootstrap/bootstrappers/litestar.py @@ -34,10 +34,8 @@ HealthChecksInstrument, HealthCheckTypedDict, ) -from microbootstrap.instruments.litestar_pydantic_descriptions import ( - PydanticDescriptionsSchemaPlugin, - PydanticTypedDictDescriptionsPlugin, -) +from microbootstrap.instruments.litestar_pydantic_descriptions import PydanticDescriptionsSchemaPlugin +from microbootstrap.instruments.litestar_typed_dict_descriptions import PydanticTypedDictDescriptionsPlugin from microbootstrap.instruments.logging_instrument import LoggingInstrument from microbootstrap.instruments.openapi_security_schemes import ( OpenApiApiKeySecurityScheme, diff --git a/microbootstrap/instruments/litestar_pydantic_descriptions.py b/microbootstrap/instruments/litestar_pydantic_descriptions.py index 486880d..22168bf 100644 --- a/microbootstrap/instruments/litestar_pydantic_descriptions.py +++ b/microbootstrap/instruments/litestar_pydantic_descriptions.py @@ -4,19 +4,18 @@ import typing import pydantic -from litestar.params import KwargDefinition, ParameterKwarg -from litestar.plugins import OpenAPISchemaPlugin +from litestar.openapi import spec +from litestar.params import KwargDefinition from litestar.plugins.pydantic import PydanticSchemaPlugin -from litestar.typing import FieldDefinition from pydantic.fields import FieldInfo if typing.TYPE_CHECKING: from litestar._openapi.schema_generation import SchemaCreator - from litestar.openapi import spec + from litestar.typing import FieldDefinition -def _fetch_field_description(field_definition: FieldDefinition) -> str | None: +def fetch_field_description(field_definition: FieldDefinition) -> str | None: keyword_definition: typing.Final = field_definition.kwarg_definition if isinstance(keyword_definition, KwargDefinition) and keyword_definition.description: return keyword_definition.description @@ -30,6 +29,19 @@ def _fetch_field_description(field_definition: FieldDefinition) -> str | None: ) +def _unwrap_root_schema(generated_schema: spec.Schema, schema_creator: SchemaCreator, root_field_name: str) -> None: + root_schema: typing.Final = (generated_schema.properties or {})[root_field_name] + value_schema: typing.Final = ( + schema_creator.schema_registry.from_reference(root_schema).schema + if isinstance(root_schema, spec.Reference) + else root_schema + ) + original_title: typing.Final = generated_schema.title + for schema_field in dataclasses.fields(spec.Schema): + setattr(generated_schema, schema_field.name, getattr(value_schema, schema_field.name)) + generated_schema.title = original_title + + @typing.final class PydanticDescriptionsSchemaPlugin(PydanticSchemaPlugin): def to_openapi_schema(self, field_definition: FieldDefinition, schema_creator: SchemaCreator) -> spec.Schema: @@ -37,41 +49,14 @@ def to_openapi_schema(self, field_definition: FieldDefinition, schema_creator: S model_type: typing.Final = field_definition.annotation if isinstance(model_type, type) and issubclass(model_type, pydantic.BaseModel): model_description: typing.Final = model_type.__dict__.get("__doc__") - if not generated_schema.description: - generated_schema.description = _fetch_field_description(field_definition) or ( - inspect.cleandoc(model_description) if isinstance(model_description, str) else None - ) + owner_description: typing.Final = fetch_field_description(field_definition) or ( + inspect.cleandoc(model_description) if isinstance(model_description, str) else None + ) + if issubclass(model_type, pydantic.RootModel): + root_field: typing.Final = model_type.model_fields["root"] + root_field_name: typing.Final = root_field.alias if schema_creator.prefer_alias else None + _unwrap_root_schema(generated_schema, schema_creator, root_field_name or "root") + generated_schema.description = owner_description or generated_schema.description + elif not generated_schema.description: + generated_schema.description = owner_description return generated_schema - - -def _describe_typed_dict_field(field_name: str, annotation: object) -> FieldDefinition: - original_field: typing.Final = FieldDefinition.from_kwarg(annotation, field_name) - original_keyword: typing.Final = original_field.kwarg_definition - description: typing.Final = _fetch_field_description(original_field) - if not description or (original_keyword is not None and not isinstance(original_keyword, KwargDefinition)): - return original_field - return dataclasses.replace( - original_field, - kwarg_definition=dataclasses.replace( - original_keyword if isinstance(original_keyword, KwargDefinition) else ParameterKwarg(), - description=description, - ), - ) - - -@typing.final -class PydanticTypedDictDescriptionsPlugin(OpenAPISchemaPlugin): - def is_plugin_supported_field(self, field_definition: FieldDefinition) -> bool: - return field_definition.is_typeddict_type - - def to_openapi_schema(self, field_definition: FieldDefinition, schema_creator: SchemaCreator) -> spec.Schema: - return schema_creator.create_component_schema( - field_definition, - required=sorted(field_definition.type_.__required_keys__), - property_fields={ - field_name: _describe_typed_dict_field(field_name, annotation) - for field_name, annotation in field_definition.get_type_hints( - include_extras=True, resolve_generics=True - ).items() - }, - ) diff --git a/microbootstrap/instruments/litestar_typed_dict_descriptions.py b/microbootstrap/instruments/litestar_typed_dict_descriptions.py new file mode 100644 index 0000000..173a148 --- /dev/null +++ b/microbootstrap/instruments/litestar_typed_dict_descriptions.py @@ -0,0 +1,47 @@ +from __future__ import annotations +import dataclasses +import typing + +from litestar.params import KwargDefinition, ParameterKwarg +from litestar.plugins import OpenAPISchemaPlugin +from litestar.typing import FieldDefinition + +from microbootstrap.instruments.litestar_pydantic_descriptions import fetch_field_description + + +if typing.TYPE_CHECKING: + from litestar._openapi.schema_generation import SchemaCreator + from litestar.openapi import spec + + +def _describe_typed_dict_field(field_name: str, annotation: object) -> FieldDefinition: + original_field: typing.Final = FieldDefinition.from_kwarg(annotation, field_name) + original_keyword: typing.Final = original_field.kwarg_definition + description: typing.Final = fetch_field_description(original_field) + if not description or (original_keyword is not None and not isinstance(original_keyword, KwargDefinition)): + return original_field + return dataclasses.replace( + original_field, + kwarg_definition=dataclasses.replace( + original_keyword if isinstance(original_keyword, KwargDefinition) else ParameterKwarg(), + description=description, + ), + ) + + +@typing.final +class PydanticTypedDictDescriptionsPlugin(OpenAPISchemaPlugin): + def is_plugin_supported_field(self, field_definition: FieldDefinition) -> bool: + return field_definition.is_typeddict_type + + def to_openapi_schema(self, field_definition: FieldDefinition, schema_creator: SchemaCreator) -> spec.Schema: + return schema_creator.create_component_schema( + field_definition, + required=sorted(field_definition.type_.__required_keys__), + property_fields={ + field_name: _describe_typed_dict_field(field_name, annotation) + for field_name, annotation in field_definition.get_type_hints( + include_extras=True, resolve_generics=True + ).items() + }, + ) diff --git a/tests/bootstrappers/test_litestar_root_models.py b/tests/bootstrappers/test_litestar_root_models.py new file mode 100644 index 0000000..f2f7e97 --- /dev/null +++ b/tests/bootstrappers/test_litestar_root_models.py @@ -0,0 +1,210 @@ +from __future__ import annotations +import http +import typing + +import litestar +import pydantic +import pytest +from litestar.plugins.pydantic import PydanticPlugin +from litestar.testing import TestClient + +from microbootstrap import LitestarSettings +from microbootstrap.bootstrappers.litestar import LitestarBootstrapper +from microbootstrap.config.litestar import LitestarConfig + + +if typing.TYPE_CHECKING: + from litestar.handlers import HTTPRouteHandler + + +_MAX_SCALAR_VALUE: typing.Final = 3 +_MAX_SEQUENCE_ITEMS: typing.Final = 3 + + +@typing.final +class ScalarRoot(pydantic.RootModel[int]): + """A bounded numeric payload.""" + + root: int = pydantic.Field(ge=1, le=_MAX_SCALAR_VALUE, description="Numeric payload.", examples=[2]) + + +@typing.final +class ArrayRoot(pydantic.RootModel[list[ScalarRoot]]): + """An ordered numeric payload.""" + + root: list[ScalarRoot] = pydantic.Field(min_length=1, max_length=_MAX_SEQUENCE_ITEMS, description="Numeric values.") + + +@typing.final +class CatalogRecord(pydantic.BaseModel): + identifier: str = pydantic.Field(description="Record identifier.") + + +@typing.final +class ObjectRoot(pydantic.RootModel[CatalogRecord]): + """A record without a wrapper.""" + + +@typing.final +class NestedRoot(pydantic.RootModel[ArrayRoot]): + """Nested numeric payload.""" + + +@typing.final +class RecursiveRoot(pydantic.RootModel[object]): + """A recursively nested sequence.""" + + root: list[RecursiveRoot] + + +@typing.final +class AliasedRoot(pydantic.RootModel[int]): + root: int = pydantic.Field(alias="payload", ge=1) + + +@typing.final +class WrappedRoots(pydantic.BaseModel): + scalar: ScalarRoot + sequence: ArrayRoot + record: ObjectRoot + nested: NestedRoot + recursive: RecursiveRoot + + +@litestar.post("/scalar", sync_to_thread=False) +def create_scalar(data: ScalarRoot) -> ScalarRoot: + return data + + +@litestar.post("/sequence", sync_to_thread=False) +def create_sequence(data: ArrayRoot) -> ArrayRoot: + return data + + +@litestar.post("/record", sync_to_thread=False) +def create_record(data: ObjectRoot) -> ObjectRoot: + return data + + +@litestar.post("/nested", sync_to_thread=False) +def create_nested(data: NestedRoot) -> NestedRoot: + return data + + +@litestar.post("/recursive", sync_to_thread=False) +def create_recursive(data: RecursiveRoot) -> RecursiveRoot: + return data + + +@litestar.post("/aliased", sync_to_thread=False) +def create_aliased(data: AliasedRoot) -> AliasedRoot: + return data + + +@litestar.post("/wrapped", sync_to_thread=False) +def create_wrapped(data: WrappedRoots) -> WrappedRoots: + return data + + +def _build_application(route_handler: HTTPRouteHandler, *, prefer_alias: bool = False) -> litestar.Litestar: + return ( + LitestarBootstrapper( + LitestarSettings(api_metadata_enabled=True, service_debug=False, health_checks_enabled=False) + ) + .configure_application( + LitestarConfig(route_handlers=[route_handler], plugins=[PydanticPlugin(prefer_alias=prefer_alias)]) + ) + .bootstrap() + ) + + +def _resolve_schema(document: dict[str, typing.Any], schema_value: dict[str, typing.Any]) -> dict[str, typing.Any]: + if "$ref" in schema_value: + return pydantic.TypeAdapter(dict[str, typing.Any]).validate_python( + document["components"]["schemas"][schema_value["$ref"].rsplit("/", maxsplit=1)[-1]] + ) + return schema_value + + +@pytest.mark.parametrize( + ("route_handler", "route_path", "request_value", "schema_type"), + [ + (create_scalar, "/scalar", 2, "integer"), + (create_sequence, "/sequence", [2], "array"), + (create_record, "/record", {"identifier": "catalog-entry"}, "object"), + (create_nested, "/nested", [2], "array"), + (create_recursive, "/recursive", [[], [[]]], "array"), + ], +) +def test_litestar_root_schema_matches_the_wire_value( + route_handler: HTTPRouteHandler, route_path: str, request_value: object, schema_type: str +) -> None: + application: typing.Final = _build_application(route_handler) + document: typing.Final = application.openapi_schema.to_schema() + operation: typing.Final = document["paths"][route_path]["post"] + request_schema: typing.Final = _resolve_schema( + document, operation["requestBody"]["content"]["application/json"]["schema"] + ) + response_schema: typing.Final = _resolve_schema( + document, operation["responses"]["201"]["content"]["application/json"]["schema"] + ) + assert request_schema["type"] == response_schema["type"] == schema_type + assert "root" not in response_schema.get("properties", {}) + with TestClient(application) as test_client: + response: typing.Final = test_client.post(route_path, json=request_value) + assert response.status_code == http.HTTPStatus.CREATED + assert response.json() == request_value + + +def test_litestar_nested_root_models_keep_constraints_and_recursive_references() -> None: + application: typing.Final = _build_application(create_wrapped) + document: typing.Final = application.openapi_schema.to_schema() + components: typing.Final = document["components"]["schemas"] + scalar_schema: typing.Final = components["ScalarRoot"] + assert scalar_schema["minimum"] == 1 + assert scalar_schema["maximum"] == _MAX_SCALAR_VALUE + assert scalar_schema["examples"] == [2] + assert scalar_schema["description"] == "A bounded numeric payload." + assert components["ArrayRoot"]["minItems"] == 1 + assert components["ArrayRoot"]["maxItems"] == _MAX_SEQUENCE_ITEMS + assert components["ObjectRoot"]["required"] == ["identifier"] + assert components["NestedRoot"]["description"] == "Nested numeric payload." + assert _resolve_schema(document, components["ArrayRoot"]["items"])["type"] == "integer" + assert components["RecursiveRoot"]["items"] == {"$ref": "#/components/schemas/RecursiveRoot"} + wrapped_schema: typing.Final = components["WrappedRoots"] + for property_name, expected_type in ( + ("scalar", "integer"), + ("sequence", "array"), + ("record", "object"), + ("nested", "array"), + ("recursive", "array"), + ): + assert _resolve_schema(document, wrapped_schema["properties"][property_name])["type"] == expected_type + request_payload: typing.Final = { + "scalar": 2, + "sequence": [2], + "record": {"identifier": "catalog-entry"}, + "nested": [2], + "recursive": [[], [[]]], + } + with TestClient(application) as test_client: + accepted: typing.Final = test_client.post("/wrapped", json=request_payload) + rejected_scalar: typing.Final = test_client.post("/wrapped", json=request_payload | {"scalar": 0}) + rejected_sequence: typing.Final = test_client.post("/wrapped", json=request_payload | {"sequence": []}) + assert accepted.status_code == http.HTTPStatus.CREATED + assert accepted.json() == request_payload + assert rejected_scalar.status_code == rejected_sequence.status_code == http.HTTPStatus.BAD_REQUEST + + +def test_litestar_root_field_alias_does_not_add_an_envelope() -> None: + application: typing.Final = _build_application(create_aliased, prefer_alias=True) + document: typing.Final = application.openapi_schema.to_schema() + root_schema: typing.Final = document["components"]["schemas"]["AliasedRoot"] + assert root_schema["type"] == "integer" + assert root_schema["minimum"] == 1 + assert "properties" not in root_schema + root_value: typing.Final = 2 + with TestClient(application) as test_client: + response: typing.Final = test_client.post("/aliased", json=root_value) + assert response.status_code == http.HTTPStatus.CREATED + assert response.json() == root_value From 044f6ca5886566c2b58054d81c090b9fb17941f4 Mon Sep 17 00:00:00 2001 From: Lev Vereshchagin Date: Mon, 2 Nov 2026 16:13:59 +0300 Subject: [PATCH 4/5] Preserve configured Pydantic schema ownership and examples --- microbootstrap/bootstrappers/litestar.py | 7 +- .../litestar_pydantic_descriptions.py | 45 ++++---- .../instruments/litestar_root_schema.py | 27 +++++ .../test_litestar_plugin_lifecycle.py | 101 ++++++++++++++++++ .../test_litestar_root_models.py | 47 ++++++-- 5 files changed, 193 insertions(+), 34 deletions(-) create mode 100644 microbootstrap/instruments/litestar_root_schema.py create mode 100644 tests/bootstrappers/test_litestar_plugin_lifecycle.py diff --git a/microbootstrap/bootstrappers/litestar.py b/microbootstrap/bootstrappers/litestar.py index 8353b57..ee1252a 100644 --- a/microbootstrap/bootstrappers/litestar.py +++ b/microbootstrap/bootstrappers/litestar.py @@ -17,7 +17,6 @@ from litestar.exceptions.responses import create_debug_response, create_exception_response from litestar.middleware import ASGIMiddleware from litestar.openapi.plugins import SwaggerRenderPlugin -from litestar.plugins.pydantic import PydanticPlugin from litestar.routes import HTTPRoute from litestar.types.asgi_types import ASGIApp, Scope from litestar_offline_docs import generate_static_files_config @@ -93,17 +92,13 @@ class LitestarBootstrapper( application_type = litestar.Litestar def bootstrap_before(self: typing_extensions.Self) -> dict[str, typing.Any]: - prefer_alias: typing.Final = next( - (plugin.prefer_alias for plugin in self.application_config.plugins if isinstance(plugin, PydanticPlugin)), - False, - ) return { "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 [], "plugins": [ - PydanticDescriptionsSchemaPlugin(prefer_alias=prefer_alias), + PydanticDescriptionsSchemaPlugin(), PydanticTypedDictDescriptionsPlugin(), ] if self.settings.api_metadata_enabled diff --git a/microbootstrap/instruments/litestar_pydantic_descriptions.py b/microbootstrap/instruments/litestar_pydantic_descriptions.py index 22168bf..2bba9af 100644 --- a/microbootstrap/instruments/litestar_pydantic_descriptions.py +++ b/microbootstrap/instruments/litestar_pydantic_descriptions.py @@ -1,17 +1,18 @@ from __future__ import annotations -import dataclasses import inspect import typing import pydantic -from litestar.openapi import spec from litestar.params import KwargDefinition from litestar.plugins.pydantic import PydanticSchemaPlugin from pydantic.fields import FieldInfo +from microbootstrap.instruments.litestar_root_schema import populate_root_value_schema + if typing.TYPE_CHECKING: from litestar._openapi.schema_generation import SchemaCreator + from litestar.openapi import spec from litestar.typing import FieldDefinition @@ -29,23 +30,25 @@ def fetch_field_description(field_definition: FieldDefinition) -> str | None: ) -def _unwrap_root_schema(generated_schema: spec.Schema, schema_creator: SchemaCreator, root_field_name: str) -> None: - root_schema: typing.Final = (generated_schema.properties or {})[root_field_name] - value_schema: typing.Final = ( - schema_creator.schema_registry.from_reference(root_schema).schema - if isinstance(root_schema, spec.Reference) - else root_schema - ) - original_title: typing.Final = generated_schema.title - for schema_field in dataclasses.fields(spec.Schema): - setattr(generated_schema, schema_field.name, getattr(value_schema, schema_field.name)) - generated_schema.title = original_title - - @typing.final class PydanticDescriptionsSchemaPlugin(PydanticSchemaPlugin): def to_openapi_schema(self, field_definition: FieldDefinition, schema_creator: SchemaCreator) -> spec.Schema: - generated_schema: typing.Final = super().to_openapi_schema(field_definition, schema_creator) + configured_plugin: typing.Final = next( + ( + plugin + for plugin in schema_creator.plugins + if not isinstance(plugin, PydanticDescriptionsSchemaPlugin) + and schema_creator.plugin_supports_field(plugin, field_definition) + ), + None, + ) + if configured_plugin is not None and type(configured_plugin) is not PydanticSchemaPlugin: + return configured_plugin.to_openapi_schema(field_definition, schema_creator) + model_schema: typing.Final = ( + configured_plugin.to_openapi_schema(field_definition, schema_creator) + if configured_plugin is not None + else super().to_openapi_schema(field_definition, schema_creator) + ) model_type: typing.Final = field_definition.annotation if isinstance(model_type, type) and issubclass(model_type, pydantic.BaseModel): model_description: typing.Final = model_type.__dict__.get("__doc__") @@ -55,8 +58,8 @@ def to_openapi_schema(self, field_definition: FieldDefinition, schema_creator: S if issubclass(model_type, pydantic.RootModel): root_field: typing.Final = model_type.model_fields["root"] root_field_name: typing.Final = root_field.alias if schema_creator.prefer_alias else None - _unwrap_root_schema(generated_schema, schema_creator, root_field_name or "root") - generated_schema.description = owner_description or generated_schema.description - elif not generated_schema.description: - generated_schema.description = owner_description - return generated_schema + populate_root_value_schema(model_schema, schema_creator, root_field_name or "root") + model_schema.description = owner_description or model_schema.description + elif not model_schema.description: + model_schema.description = owner_description + return model_schema diff --git a/microbootstrap/instruments/litestar_root_schema.py b/microbootstrap/instruments/litestar_root_schema.py new file mode 100644 index 0000000..0378d43 --- /dev/null +++ b/microbootstrap/instruments/litestar_root_schema.py @@ -0,0 +1,27 @@ +from __future__ import annotations +import dataclasses +import typing + +from litestar.openapi import spec + + +if typing.TYPE_CHECKING: + from litestar._openapi.schema_generation import SchemaCreator + + +def populate_root_value_schema( + generated_schema: spec.Schema, schema_creator: SchemaCreator, root_field_name: str +) -> None: + root_schema: typing.Final = (generated_schema.properties or {})[root_field_name] + value_schema: typing.Final = ( + schema_creator.schema_registry.from_reference(root_schema).schema + if isinstance(root_schema, spec.Reference) + else root_schema + ) + original_title: typing.Final = generated_schema.title + original_examples: typing.Final = generated_schema.examples + for schema_field in dataclasses.fields(spec.Schema): + setattr(generated_schema, schema_field.name, getattr(value_schema, schema_field.name)) + generated_schema.title = original_title + if original_examples is not None: + generated_schema.examples = original_examples diff --git a/tests/bootstrappers/test_litestar_plugin_lifecycle.py b/tests/bootstrappers/test_litestar_plugin_lifecycle.py new file mode 100644 index 0000000..8046fc2 --- /dev/null +++ b/tests/bootstrappers/test_litestar_plugin_lifecycle.py @@ -0,0 +1,101 @@ +from __future__ import annotations +import typing + +import litestar +import pydantic +import pytest +from litestar.plugins import InitPlugin +from litestar.plugins.pydantic import PydanticPlugin, PydanticSchemaPlugin +from litestar.testing import TestClient + +from microbootstrap import LitestarSettings +from microbootstrap.bootstrappers.litestar import LitestarBootstrapper +from microbootstrap.config.litestar import LitestarConfig +from microbootstrap.instruments.base import BaseInstrumentConfig, Instrument +from microbootstrap.instruments.instrument_box import InstrumentBox + + +if typing.TYPE_CHECKING: + from litestar._openapi.schema_generation import SchemaCreator + from litestar.config.app import AppConfig + from litestar.openapi import spec + from litestar.plugins import PluginProtocol + from litestar.typing import FieldDefinition + + +@typing.final +class CatalogIdentity(pydantic.BaseModel): + identity: str = pydantic.Field(alias="publicIdentity", description="Public catalog identity.") + + +def _build_application( + bootstrapper_type: type[LitestarBootstrapper], + *, + metadata_enabled: bool, + plugins: list[PluginProtocol] | None = None, +) -> litestar.Litestar: + @litestar.get("/identity", sync_to_thread=False) + def fetch_identity() -> CatalogIdentity: + return CatalogIdentity(publicIdentity="catalog-entry") + + return ( + bootstrapper_type( + LitestarSettings(api_metadata_enabled=metadata_enabled, service_debug=False, health_checks_enabled=False) + ) + .configure_application(LitestarConfig(route_handlers=[fetch_identity], plugins=plugins or [])) + .bootstrap() + ) + + +@pytest.mark.parametrize("metadata_enabled", [False, True]) +def test_litestar_instrument_alias_policy_matches_the_wire_value(metadata_enabled: bool) -> None: + @typing.final + class AliasBootstrapper(LitestarBootstrapper): + instrument_box = InstrumentBox() + + @AliasBootstrapper.use_instrument() + @typing.final + class AliasInstrument(Instrument[BaseInstrumentConfig]): + instrument_name = "Pydantic alias policy" + ready_condition = "Always enabled for the catalog" + + @classmethod + def get_config_type(cls) -> type[BaseInstrumentConfig]: + return BaseInstrumentConfig + + def is_ready(self) -> bool: + return True + + def bootstrap_before(self) -> dict[str, list[PluginProtocol]]: + return {"plugins": [PydanticPlugin(prefer_alias=True)]} + + application: typing.Final = _build_application(AliasBootstrapper, metadata_enabled=metadata_enabled) + document: typing.Final = application.openapi_schema.to_schema() + catalog_schema: typing.Final = document["components"]["schemas"]["CatalogIdentity"] + assert set(catalog_schema["properties"]) == {"publicIdentity"} + with TestClient(application) as test_client: + assert test_client.get("/identity").json() == {"publicIdentity": "catalog-entry"} + + +@pytest.mark.parametrize("metadata_enabled", [False, True]) +def test_litestar_init_plugin_keeps_ownership_of_schema_generation(metadata_enabled: bool) -> None: + @typing.final + class CatalogSchemaPlugin(PydanticSchemaPlugin): + def to_openapi_schema(self, field_definition: FieldDefinition, schema_creator: SchemaCreator) -> spec.Schema: + generated_schema: typing.Final = super().to_openapi_schema(field_definition, schema_creator) + generated_schema.description = "Application-owned catalog contract." + return generated_schema + + @typing.final + class CatalogInitPlugin(InitPlugin): + def on_app_init(self, app_config: AppConfig) -> AppConfig: + app_config.plugins.append(CatalogSchemaPlugin()) + return app_config + + application: typing.Final = _build_application( + LitestarBootstrapper, metadata_enabled=metadata_enabled, plugins=[CatalogInitPlugin()] + ) + document: typing.Final = application.openapi_schema.to_schema() + assert document["components"]["schemas"]["CatalogIdentity"]["description"] == "Application-owned catalog contract." + with TestClient(application) as test_client: + assert test_client.get("/identity").json() == {"identity": "catalog-entry"} diff --git a/tests/bootstrappers/test_litestar_root_models.py b/tests/bootstrappers/test_litestar_root_models.py index f2f7e97..42edda4 100644 --- a/tests/bootstrappers/test_litestar_root_models.py +++ b/tests/bootstrappers/test_litestar_root_models.py @@ -28,6 +28,16 @@ class ScalarRoot(pydantic.RootModel[int]): root: int = pydantic.Field(ge=1, le=_MAX_SCALAR_VALUE, description="Numeric payload.", examples=[2]) +class RootExampleConfig(pydantic.ConfigDict): + example: int + + +@typing.final +class ConfiguredExamplesRoot(pydantic.RootModel[int]): + model_config: typing.ClassVar[RootExampleConfig] = {"example": 3} + root: int = pydantic.Field(ge=1, le=_MAX_SCALAR_VALUE, examples=[2]) + + @typing.final class ArrayRoot(pydantic.RootModel[list[ScalarRoot]]): """An ordered numeric payload.""" @@ -76,6 +86,11 @@ def create_scalar(data: ScalarRoot) -> ScalarRoot: return data +@litestar.post("/configured", sync_to_thread=False) +def create_configured(data: ConfiguredExamplesRoot) -> ConfiguredExamplesRoot: + return data + + @litestar.post("/sequence", sync_to_thread=False) def create_sequence(data: ArrayRoot) -> ArrayRoot: return data @@ -118,12 +133,16 @@ def _build_application(route_handler: HTTPRouteHandler, *, prefer_alias: bool = ) -def _resolve_schema(document: dict[str, typing.Any], schema_value: dict[str, typing.Any]) -> dict[str, typing.Any]: - if "$ref" in schema_value: - return pydantic.TypeAdapter(dict[str, typing.Any]).validate_python( - document["components"]["schemas"][schema_value["$ref"].rsplit("/", maxsplit=1)[-1]] - ) - return schema_value +def _resolve_schema( + document: dict[str, pydantic.JsonValue], schema_value: dict[str, pydantic.JsonValue] +) -> dict[str, pydantic.JsonValue]: + reference_value: typing.Final = schema_value.get("$ref") + if not isinstance(reference_value, str): + return schema_value + object_adapter: typing.Final = pydantic.TypeAdapter(dict[str, pydantic.JsonValue]) + components: typing.Final = object_adapter.validate_python(document.get("components")) + schemas: typing.Final = object_adapter.validate_python(components.get("schemas")) + return object_adapter.validate_python(schemas.get(reference_value.rsplit("/", maxsplit=1)[-1])) @pytest.mark.parametrize( @@ -149,7 +168,8 @@ def test_litestar_root_schema_matches_the_wire_value( document, operation["responses"]["201"]["content"]["application/json"]["schema"] ) assert request_schema["type"] == response_schema["type"] == schema_type - assert "root" not in response_schema.get("properties", {}) + response_properties: typing.Final = response_schema.get("properties") + assert not isinstance(response_properties, dict) or "root" not in response_properties with TestClient(application) as test_client: response: typing.Final = test_client.post(route_path, json=request_value) assert response.status_code == http.HTTPStatus.CREATED @@ -208,3 +228,16 @@ def test_litestar_root_field_alias_does_not_add_an_envelope() -> None: response: typing.Final = test_client.post("/aliased", json=root_value) assert response.status_code == http.HTTPStatus.CREATED assert response.json() == root_value + + +def test_litestar_root_model_examples_override_root_field_examples() -> None: + application: typing.Final = _build_application(create_configured) + document: typing.Final = application.openapi_schema.to_schema() + root_schema: typing.Final = document["components"]["schemas"]["ConfiguredExamplesRoot"] + configured_example: typing.Final = 3 + assert root_schema["examples"] == [configured_example] + assert root_schema["type"] == "integer" + with TestClient(application) as test_client: + response: typing.Final = test_client.post("/configured", json=configured_example) + assert response.status_code == http.HTTPStatus.CREATED + assert response.json() == configured_example From eacc698c87d9c7903fb6a8a7d31b9f8784656c5c Mon Sep 17 00:00:00 2001 From: Lev Vereshchagin Date: Mon, 2 Nov 2026 16:32:22 +0300 Subject: [PATCH 5/5] Preserve native use-site examples for root schemas --- .../litestar_pydantic_descriptions.py | 2 +- .../instruments/litestar_root_schema.py | 12 +++++-- .../test_litestar_root_models.py | 34 +++++++++++++++++++ 3 files changed, 45 insertions(+), 3 deletions(-) diff --git a/microbootstrap/instruments/litestar_pydantic_descriptions.py b/microbootstrap/instruments/litestar_pydantic_descriptions.py index 2bba9af..97d3e11 100644 --- a/microbootstrap/instruments/litestar_pydantic_descriptions.py +++ b/microbootstrap/instruments/litestar_pydantic_descriptions.py @@ -58,7 +58,7 @@ def to_openapi_schema(self, field_definition: FieldDefinition, schema_creator: S if issubclass(model_type, pydantic.RootModel): root_field: typing.Final = model_type.model_fields["root"] root_field_name: typing.Final = root_field.alias if schema_creator.prefer_alias else None - populate_root_value_schema(model_schema, schema_creator, root_field_name or "root") + populate_root_value_schema(model_schema, schema_creator, root_field_name or "root", field_definition) model_schema.description = owner_description or model_schema.description elif not model_schema.description: model_schema.description = owner_description diff --git a/microbootstrap/instruments/litestar_root_schema.py b/microbootstrap/instruments/litestar_root_schema.py index 0378d43..b4ed492 100644 --- a/microbootstrap/instruments/litestar_root_schema.py +++ b/microbootstrap/instruments/litestar_root_schema.py @@ -3,14 +3,19 @@ import typing from litestar.openapi import spec +from litestar.params import KwargDefinition if typing.TYPE_CHECKING: from litestar._openapi.schema_generation import SchemaCreator + from litestar.typing import FieldDefinition def populate_root_value_schema( - generated_schema: spec.Schema, schema_creator: SchemaCreator, root_field_name: str + generated_schema: spec.Schema, + schema_creator: SchemaCreator, + root_field_name: str, + field_definition: FieldDefinition, ) -> None: root_schema: typing.Final = (generated_schema.properties or {})[root_field_name] value_schema: typing.Final = ( @@ -23,5 +28,8 @@ def populate_root_value_schema( for schema_field in dataclasses.fields(spec.Schema): setattr(generated_schema, schema_field.name, getattr(value_schema, schema_field.name)) generated_schema.title = original_title - if original_examples is not None: + keyword_definition: typing.Final = field_definition.kwarg_definition + if original_examples is not None or ( + isinstance(keyword_definition, KwargDefinition) and keyword_definition.examples + ): generated_schema.examples = original_examples diff --git a/tests/bootstrappers/test_litestar_root_models.py b/tests/bootstrappers/test_litestar_root_models.py index 42edda4..934ad20 100644 --- a/tests/bootstrappers/test_litestar_root_models.py +++ b/tests/bootstrappers/test_litestar_root_models.py @@ -5,6 +5,8 @@ import litestar import pydantic import pytest +from litestar.openapi.spec import Example +from litestar.params import Parameter from litestar.plugins.pydantic import PydanticPlugin from litestar.testing import TestClient @@ -19,6 +21,7 @@ _MAX_SCALAR_VALUE: typing.Final = 3 _MAX_SEQUENCE_ITEMS: typing.Final = 3 +_USE_SITE_EXAMPLE: typing.Final = 3 @typing.final @@ -241,3 +244,34 @@ def test_litestar_root_model_examples_override_root_field_examples() -> None: response: typing.Final = test_client.post("/configured", json=configured_example) assert response.status_code == http.HTTPStatus.CREATED assert response.json() == configured_example + + +@litestar.get("/catalog", sync_to_thread=False) +def fetch_catalog() -> typing.Annotated[ScalarRoot, Parameter(examples=[Example(value=_USE_SITE_EXAMPLE)])]: + return ScalarRoot(_USE_SITE_EXAMPLE) + + +@litestar.get("/configured-catalog", sync_to_thread=False) +def fetch_configured_catalog() -> typing.Annotated[ConfiguredExamplesRoot, Parameter(examples=[Example(value=1)])]: + return ConfiguredExamplesRoot(_USE_SITE_EXAMPLE) + + +@pytest.mark.parametrize( + ("route_handler", "route_path", "component_name"), + [ + (fetch_catalog, "/catalog", "ScalarRoot"), + (fetch_configured_catalog, "/configured-catalog", "ConfiguredExamplesRoot"), + ], +) +def test_litestar_root_examples_keep_native_use_site_precedence( + route_handler: HTTPRouteHandler, route_path: str, component_name: str +) -> None: + application: typing.Final = _build_application(route_handler) + document: typing.Final = application.openapi_schema.to_schema() + root_schema: typing.Final = document["components"]["schemas"][component_name] + assert root_schema["type"] == "integer" + assert root_schema["examples"] == [_USE_SITE_EXAMPLE] + with TestClient(application) as test_client: + response: typing.Final = test_client.get(route_path) + assert response.status_code == http.HTTPStatus.OK + assert response.json() == _USE_SITE_EXAMPLE