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
29 changes: 17 additions & 12 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -730,26 +730,31 @@ 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
### API metadata

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.
Set `api_metadata_enabled=True` in `FastApiSettings` or `LitestarSettings` to enable
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.

```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.
app = FastApiBootstrapper(FastApiSettings(api_metadata_enabled=True)).bootstrap()
```

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.
Default HTTP errors receive the HTTP status phrase as `title`, alongside the
framework's existing fields, status and headers. Generated validation schemas
and examples reflect that response. Application, router and route exception
handlers retain ownership of their payloads and schemas. Debug tracebacks retain
the framework's native behavior. Error titles also work with Swagger disabled.
This setting does not document authentication, change successful responses or
certify compliance with an API standard.

`api_metadata_enabled` replaces the description-only `openapi_descriptions_enabled`
option from 1.8.0. Migrate existing settings and environment variables to the new
name; enabling it also changes default HTTP error bodies as described above.

## Configuration

Expand Down
111 changes: 88 additions & 23 deletions microbootstrap/bootstrappers/fastapi.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,20 @@
import contextlib
import http
import typing

import fastapi
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi_offline_docs import enable_offline_docs
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from prometheus_fastapi_instrumentator import Instrumentator, metrics
from starlette.exceptions import HTTPException
from starlette.responses import JSONResponse

from microbootstrap.bootstrappers.base import ApplicationBootstrapper
from microbootstrap.config.fastapi import FastApiConfig
from microbootstrap.helpers import http_error_title
from microbootstrap.instruments.cors_instrument import CorsInstrument
from microbootstrap.instruments.health_checks_instrument import HealthChecksInstrument, HealthCheckTypedDict
from microbootstrap.instruments.logging_instrument import LoggingInstrument
Expand Down Expand Up @@ -62,6 +68,87 @@ def bootstrap_before(self) -> dict[str, typing.Any]:
"lifespan": self._wrapped_lifespan_manager if self.application_config.lifespan else self._lifespan_manager,
}

def bootstrap_before_instruments_after_app_created(self, application: ApplicationT) -> ApplicationT:
if self.settings.api_metadata_enabled:
for exception, native, replacement in (
(HTTPException, fastapi.exception_handlers.http_exception_handler, self.http_exception_response),
(
RequestValidationError,
fastapi.exception_handlers.request_validation_exception_handler,
self.validation_error_response,
),
):
if application.exception_handlers.get(exception) == native:
application.exception_handlers[exception] = replacement
if http.HTTPStatus.INTERNAL_SERVER_ERROR not in application.exception_handlers:
application.exception_handlers.setdefault(Exception, self.internal_error_response)
return application

async def http_exception_response(self, request: fastapi.Request, exc: HTTPException) -> fastapi.Response:
response = await fastapi.exception_handlers.http_exception_handler(request, exc)
if exc.status_code >= http.HTTPStatus.BAD_REQUEST:
return JSONResponse(
{"detail": exc.detail, "title": http_error_title(exc.status_code)},
status_code=exc.status_code,
headers=exc.headers,
)
return response

async def validation_error_response(self, _request: fastapi.Request, exc: RequestValidationError) -> JSONResponse:
return JSONResponse(
{
"detail": jsonable_encoder(exc.errors()),
"title": http_error_title(http.HTTPStatus.UNPROCESSABLE_ENTITY),
},
status_code=http.HTTPStatus.UNPROCESSABLE_ENTITY,
)

async def internal_error_response(self, _request: fastapi.Request, _exc: Exception) -> JSONResponse:
return JSONResponse(
{"detail": "Internal Server Error", "title": http_error_title(http.HTTPStatus.INTERNAL_SERVER_ERROR)},
status_code=http.HTTPStatus.INTERNAL_SERVER_ERROR,
)

def bootstrap_after(self, application: ApplicationT) -> ApplicationT:
if not self.settings.api_metadata_enabled:
return application
original_openapi: typing.Final = application.openapi

def titled_openapi() -> dict[str, typing.Any]:
document: typing.Final = original_openapi()
self._describe_generated_schemas(document)
schema = document.get("components", {}).get("schemas", {}).get("HTTPValidationError")
if (
schema is not None
and application.exception_handlers.get(RequestValidationError) == self.validation_error_response
):
schema.setdefault("properties", {}).setdefault(
"title", {"type": "string", "description": "Short, human-readable summary of the HTTP error."}
)
required = schema.setdefault("required", [])
if "title" not in required:
required.append("title")
return document

application.openapi = titled_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."
)


FastApiBootstrapper.use_instrument()(SentryInstrument)

Expand All @@ -79,19 +166,13 @@ 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
and not self.instrument_config.openapi_descriptions_enabled
):
if self.instrument_config.openapi_version_docs is None and not self.instrument_config.security_schemes:
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 @@ -101,22 +182,6 @@ 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
Loading
Loading