diff --git a/README.md b/README.md index 12850d9..b86ae6b 100644 --- a/README.md +++ b/README.md @@ -737,6 +737,13 @@ 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. +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 from microbootstrap.bootstrappers.fastapi import FastApiBootstrapper diff --git a/microbootstrap/bootstrappers/litestar.py b/microbootstrap/bootstrappers/litestar.py index 67e9d16..ee1252a 100644 --- a/microbootstrap/bootstrappers/litestar.py +++ b/microbootstrap/bootstrappers/litestar.py @@ -33,6 +33,8 @@ HealthChecksInstrument, HealthCheckTypedDict, ) +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, @@ -95,6 +97,12 @@ def bootstrap_before(self: typing_extensions.Self) -> dict[str, typing.Any]: "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(), + PydanticTypedDictDescriptionsPlugin(), + ] + if self.settings.api_metadata_enabled + else [], } def exception_response( @@ -472,7 +480,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/microbootstrap/instruments/litestar_pydantic_descriptions.py b/microbootstrap/instruments/litestar_pydantic_descriptions.py new file mode 100644 index 0000000..97d3e11 --- /dev/null +++ b/microbootstrap/instruments/litestar_pydantic_descriptions.py @@ -0,0 +1,65 @@ +from __future__ import annotations +import inspect +import typing + +import pydantic +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 + + +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: + 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__") + 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 + 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 + 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..b4ed492 --- /dev/null +++ b/microbootstrap/instruments/litestar_root_schema.py @@ -0,0 +1,35 @@ +from __future__ import annotations +import dataclasses +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, + field_definition: FieldDefinition, +) -> 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 + 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/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_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_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"} diff --git a/tests/bootstrappers/test_litestar_root_models.py b/tests/bootstrappers/test_litestar_root_models.py new file mode 100644 index 0000000..934ad20 --- /dev/null +++ b/tests/bootstrappers/test_litestar_root_models.py @@ -0,0 +1,277 @@ +from __future__ import annotations +import http +import typing + +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 + +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 +_USE_SITE_EXAMPLE: 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]) + + +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.""" + + 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("/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 + + +@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, 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( + ("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 + 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 + 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 + + +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 + + +@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 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: