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
21 changes: 21 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -730,6 +730,27 @@ Parameter descriptions:
- `health_checks_path` - Path for health check handler.
- `health_checks_include_in_schema` - Must be True to include `health_checks_path` (`/health/`) in OpenAPI schema.

### Generated OpenAPI descriptions

Set `openapi_descriptions_enabled=True` in `FastApiSettings`, `LitestarSettings`,
or `SwaggerConfig` to describe framework-generated validation errors and JSON
values automatically. Litestar also fills descriptions for microbootstrap health
fields. The setting is disabled by default and supports the normal environment
prefix, for example `OPENAPI_DESCRIPTIONS_ENABLED=true` without a prefix.

```python
from microbootstrap import FastApiSettings
from microbootstrap.bootstrappers.fastapi import FastApiBootstrapper

app = FastApiBootstrapper(FastApiSettings(openapi_descriptions_enabled=True)).bootstrap()
# Register application routes normally; FastAPI enriches its schema when generated.
```

The bootstrappers preserve existing descriptions and runtime response shapes.
Application-specific descriptions and authentication declarations remain with the
application, using native framework metadata and the existing `security_schemes`
setting. No authentication policy is inferred or enforced by this option.

## Configuration

While settings provide a convenient mechanism, it's not always feasible to store everything within them.
Expand Down
34 changes: 33 additions & 1 deletion microbootstrap/bootstrappers/fastapi.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,16 @@
from microbootstrap.settings import FastApiSettings


_VALIDATION_DESCRIPTIONS: typing.Final = {
"detail": "Validation errors describing the rejected request fields.",
"loc": "Field names and array indices locating the invalid input.",
"msg": "Human-readable validation error message.",
"type": "Machine-readable validation error category.",
"input": "Input value associated with the validation failure.",
"ctx": "Additional context explaining the validation error.",
}


ApplicationT = typing.TypeVar("ApplicationT", bound=fastapi.FastAPI)


Expand Down Expand Up @@ -69,13 +79,19 @@ def bootstrap_before(self) -> dict[str, typing.Any]:
def bootstrap_after(self, application: ApplicationT) -> ApplicationT:
if self.instrument_config.swagger_offline_docs:
enable_offline_docs(application, static_files_handler=self.instrument_config.service_static_path)
if self.instrument_config.openapi_version_docs is None and not self.instrument_config.security_schemes:
if (
self.instrument_config.openapi_version_docs is None
and not self.instrument_config.security_schemes
and not self.instrument_config.openapi_descriptions_enabled
):
return application

original_openapi: typing.Final = application.openapi

def documented_openapi() -> dict[str, typing.Any]:
openapi_schema: typing.Final = original_openapi()
if self.instrument_config.openapi_descriptions_enabled:
self._describe_generated_schemas(openapi_schema)
if self.instrument_config.security_schemes:
self._merge_security_schemes(openapi_schema)
if self.instrument_config.openapi_version_docs is not None:
Expand All @@ -85,6 +101,22 @@ def documented_openapi() -> dict[str, typing.Any]:
application.openapi = documented_openapi # type: ignore[method-assign] # FastAPI's public custom OpenAPI hook.
return application

@staticmethod
def _describe_generated_schemas(document: dict[str, typing.Any]) -> None:
schemas = document.get("components", {}).get("schemas", {})
for name in ("HTTPValidationError", "ValidationError"):
if name not in schemas:
continue
definition = schemas[name]
definition.setdefault("description", "Request validation failure reported by FastAPI.")
for field_name, description in _VALIDATION_DESCRIPTIONS.items():
if field_name in definition.get("properties", {}):
definition["properties"][field_name].setdefault("description", description)
if "JsonValue" in schemas:
schemas["JsonValue"].setdefault(
"description", "JSON value: an object, array, string, number, boolean or null."
)

def _merge_security_schemes(self, openapi_schema: dict[str, typing.Any]) -> None:
expected_schemes: typing.Final = serialize_security_schemes(self.instrument_config.security_schemes)
components = openapi_schema.setdefault("components", {})
Expand Down
64 changes: 61 additions & 3 deletions microbootstrap/bootstrappers/litestar.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import typing

import litestar
import pydantic
import typing_extensions
from litestar import openapi
from litestar.config.cors import CORSConfig as LitestarCorsConfig
Expand Down Expand Up @@ -47,6 +48,17 @@
from microbootstrap.settings import LitestarSettings


_ERROR_DESCRIPTIONS: typing.Final = {
"status_code": "HTTP status code of the failed request.",
"detail": "Human-readable explanation of the request failure.",
"extra": "Additional validation or error details supplied by Litestar.",
}
_HEALTH_PROPERTIES: typing.Final[dict[str, str]] = {
name: field["description"]
for name, field in pydantic.TypeAdapter(HealthCheckTypedDict).json_schema()["properties"].items()
}


ApplicationT = typing.TypeVar("ApplicationT", bound=litestar.Litestar)


Expand Down Expand Up @@ -124,16 +136,62 @@ def bootstrap_before(self) -> dict[str, typing.Any]:
return bootstrap_result

def bootstrap_after(self, application: ApplicationT) -> ApplicationT:
if (self.instrument_config.openapi_version_docs is None and not self.instrument_config.security_schemes) or (
application.openapi_config is None
):
if (
self.instrument_config.openapi_version_docs is None
and not self.instrument_config.security_schemes
and not self.instrument_config.openapi_descriptions_enabled
) or (application.openapi_config is None):
return application
if self.instrument_config.openapi_descriptions_enabled:
self._describe_generated_schemas(application.openapi_schema)
if self.instrument_config.security_schemes:
self._merge_security_schemes(application.openapi_schema)
if self.instrument_config.openapi_version_docs is not None:
self._document_operations(application.openapi_schema)
return application

@staticmethod
def _describe_properties(schema: openapi.spec.Schema, descriptions: typing.Mapping[str, str]) -> None:
for name, description in descriptions.items():
field = (schema.properties or {}).get(name)
if field is not None and not field.description:
field.description = description

@classmethod
def _describe_error_schema(cls, schema: openapi.spec.Schema) -> None:
for branch in (*(schema.one_of or []), *(schema.any_of or []), *(schema.all_of or [])):
if isinstance(branch, openapi.spec.Schema):
cls._describe_error_schema(branch)
if (
schema.type is None
and schema.one_of
and all(
isinstance(branch, openapi.spec.Schema) and branch.type == openapi.spec.OpenAPIType.OBJECT
for branch in schema.one_of
)
):
schema.type = openapi.spec.OpenAPIType.OBJECT
cls._describe_properties(schema, _ERROR_DESCRIPTIONS)

@classmethod
def _describe_generated_schemas(cls, document: openapi.spec.OpenAPI) -> None:
schemas = document.components.schemas or {}
if "JsonValue" in schemas and not schemas["JsonValue"].description:
schemas["JsonValue"].description = "JSON value: an object, array, string, number, boolean or null."
if "HealthCheckTypedDict" in schemas:
cls._describe_properties(schemas["HealthCheckTypedDict"], _HEALTH_PROPERTIES)
for path_item in (document.paths or {}).values():
for method in SUPPORTED_HTTP_METHODS:
operation = getattr(path_item, method)
if operation is None:
continue
for code, response in (operation.responses or {}).items():
if not str(code).startswith(("4", "5")) or not isinstance(response, openapi.spec.OpenAPIResponse):
continue
for media_type in (response.content or {}).values():
if isinstance(media_type.schema, openapi.spec.Schema):
cls._describe_error_schema(media_type.schema)

def _merge_security_schemes(self, openapi_schema: openapi.spec.OpenAPI) -> None:
expected_schemes: typing.Final = serialize_security_schemes(self.instrument_config.security_schemes)
security_schemes = openapi_schema.components.security_schemes
Expand Down
11 changes: 8 additions & 3 deletions microbootstrap/instruments/health_checks_instrument.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,19 @@
from __future__ import annotations
import typing

import pydantic
import typing_extensions

from microbootstrap.instruments.base import BaseInstrumentConfig, Instrument


class HealthCheckTypedDict(typing_extensions.TypedDict, total=False):
service_version: str | None
service_name: str | None
health_status: bool
service_version: typing.Annotated[str | None, pydantic.Field(description="Version of the running service.")]
service_name: typing.Annotated[str | None, pydantic.Field(description="Name of the running service.")]
health_status: typing.Annotated[
bool,
pydantic.Field(description="Static process health status; external dependencies are not checked."),
]


class HealthChecksConfig(BaseInstrumentConfig):
Expand Down
1 change: 1 addition & 0 deletions microbootstrap/instruments/swagger_instrument.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ class SwaggerConfig(BaseInstrumentConfig):
swagger_extra_params: dict[str, typing.Any] = pydantic.Field(default_factory=dict)
security_schemes: typing.Mapping[str, _OpenApiSecurityScheme] = pydantic.Field(default_factory=dict)
openapi_version_docs: OpenApiVersionDocsConfig | None = None
openapi_descriptions_enabled: bool = False

@pydantic.field_validator("security_schemes")
@classmethod
Expand Down
Loading
Loading