Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
17 changes: 16 additions & 1 deletion microbootstrap/bootstrappers/litestar.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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()

Expand Down
65 changes: 65 additions & 0 deletions microbootstrap/instruments/litestar_pydantic_descriptions.py
Original file line number Diff line number Diff line change
@@ -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
35 changes: 35 additions & 0 deletions microbootstrap/instruments/litestar_root_schema.py
Original file line number Diff line number Diff line change
@@ -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
47 changes: 47 additions & 0 deletions microbootstrap/instruments/litestar_typed_dict_descriptions.py
Original file line number Diff line number Diff line change
@@ -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()
},
)
101 changes: 101 additions & 0 deletions tests/bootstrappers/test_litestar_plugin_lifecycle.py
Original file line number Diff line number Diff line change
@@ -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"}
Loading
Loading