Add configurable OpenAPI security schemes and API version documentation

README.md

@@ -40,7 +40,7 @@ With <b>microbootstrap</b>, you receive an application with lightweight built-in
 - `opentelemetry`
 - `logging`
 - `cors`
-- `swagger` - with additional offline version support
+- `swagger` - offline UI assets, OpenAPI security definitions, and optional Accept-version documentation
 - `health-checks`
 
 Those instruments can be bootstrapped for:
@@ -594,6 +594,64 @@ Parameter descriptions:
 - `swagger_offline_docs` - A boolean value that, when set to True, allows the Swagger JS bundles to be accessed offline. This is because the service starts to host via static.
 - `swagger_extra_params` - Additional parameters to pass into the OpenAPI configuration.
 
+#### OpenAPI security schemes
+
+Security schemes are disabled by default. They add reusable OpenAPI definitions under
+`components.securitySchemes`; they do not authenticate requests or add global or operation-level security requirements.
+Keep requirements and authentication in your application routes and dependencies.
+
+```python
+from microbootstrap import (
+    LitestarSettings,
+    OpenApiHttpSecurityScheme,
+)
+
+
+class Settings(LitestarSettings):
+    security_schemes: dict[str, OpenApiHttpSecurityScheme] = {
+        "serviceAuth": OpenApiHttpSecurityScheme(scheme="bearer", bearer_format="JWT"),
+    }
+```
+
+HTTP, API key, OAuth 2.0, and OpenID Connect definitions are supported by `SwaggerConfig`. Annotating a consumer
+setting with a concrete scheme class intentionally rejects other kinds for that consumer. Python field names and
+OpenAPI aliases are accepted; output uses canonical names such as `bearerFormat`, `in`, `tokenUrl`, and
+`openIdConnectUrl`. A same-named definition must be identical to the service-owned definition or schema generation
+raises `ValueError`.
+
+The definitions are added to the framework's normal schema. Repeated schema reads remain stable; after correcting
+a configuration conflict, rebuild the application.
+
+#### API-version documentation
+
+Version documentation is disabled by default. It describes supported versions in OpenAPI without implementing
+runtime version negotiation.
+
+```python
+from microbootstrap import LitestarSettings, OpenApiOperationVersionOverride, OpenApiVersionDocsConfig
+
+
+class Settings(LitestarSettings):
+    openapi_version_docs: OpenApiVersionDocsConfig | None = OpenApiVersionDocsConfig(
+        vendor_media_type="application/vnd.example+json",
+        supported_versions=("1.0",),
+        operation_versions=(
+            OpenApiOperationVersionOverride(path="/widgets", method="post", supported_versions=("2.0",)),
+            OpenApiOperationVersionOverride(path="/internal/widgets", method="get", supported_versions=()),
+        ),
+    )
+```
+
+Set `openapi_version_docs` to `None` to disable version documentation. A configured non-empty global
+`supported_versions` list adds an `x-accept-versioning` extension and matching description text to each operation.
+`operation_versions` replaces that list for one exact path and lower-case HTTP method; an explicit empty tuple skips
+microbootstrap's additions for that operation without asserting that no service-owned version metadata exists. This does
+not negotiate requests, add an `Accept` parameter, change response media types, or provide a Swagger UI version selector.
+
+Repeated schema reads remain stable. A conflicting service-owned `x-accept-versioning` extension raises `ValueError`;
+correct the configuration and rebuild the application. Litestar supports its standard `Operation` type for version
+documentation and rejects other custom operation subclasses.
+
 #### FastStream AsyncAPI documentation
 
 AsyncAPI documentation is available by default under `/asyncapi` route. You can change that by setting `asyncapi_path`:

microbootstrap/__init__.py

@@ -1,6 +1,18 @@
 from microbootstrap.instruments.cors_instrument import CorsConfig
 from microbootstrap.instruments.health_checks_instrument import HealthChecksConfig
 from microbootstrap.instruments.logging_instrument import LoggingConfig
+from microbootstrap.instruments.openapi_security_schemes import (
+    OpenApiApiKeySecurityScheme,
+    OpenApiHttpSecurityScheme,
+    OpenApiOAuth2SecurityScheme,
+    OpenApiOAuthFlow,
+    OpenApiOAuthFlows,
+    OpenApiOpenIdConnectSecurityScheme,
+)
+from microbootstrap.instruments.openapi_version_docs import (
+    OpenApiOperationVersionOverride,
+    OpenApiVersionDocsConfig,
+)
 from microbootstrap.instruments.opentelemetry_instrument import (
     FastStreamOpentelemetryConfig,
     FastStreamTelemetryMiddlewareProtocol,
@@ -42,6 +54,14 @@
     "LitestarPrometheusConfig",
     "LitestarSettings",
     "LoggingConfig",
+    "OpenApiApiKeySecurityScheme",
+    "OpenApiHttpSecurityScheme",
+    "OpenApiOAuth2SecurityScheme",
+    "OpenApiOAuthFlow",
+    "OpenApiOAuthFlows",
+    "OpenApiOpenIdConnectSecurityScheme",
+    "OpenApiOperationVersionOverride",
+    "OpenApiVersionDocsConfig",
     "OpentelemetryConfig",
     "PyroscopeConfig",
     "SentryConfig",

microbootstrap/bootstrappers/fastapi.py

@@ -12,6 +12,8 @@
 from microbootstrap.instruments.cors_instrument import CorsInstrument
 from microbootstrap.instruments.health_checks_instrument import HealthChecksInstrument, HealthCheckTypedDict
 from microbootstrap.instruments.logging_instrument import LoggingInstrument
+from microbootstrap.instruments.openapi_security_schemes import serialize_security_schemes
+from microbootstrap.instruments.openapi_version_docs import SUPPORTED_HTTP_METHODS
 from microbootstrap.instruments.opentelemetry_instrument import OpentelemetryInstrument
 from microbootstrap.instruments.prometheus_instrument import FastApiPrometheusConfig, PrometheusInstrument
 from microbootstrap.instruments.pyroscope_instrument import PyroscopeInstrument
@@ -67,8 +69,48 @@ 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:
+            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.security_schemes:
+                self._merge_security_schemes(openapi_schema)
+            if self.instrument_config.openapi_version_docs is not None:
+                self._document_operations(openapi_schema)
+            return openapi_schema
+
+        application.openapi = documented_openapi  # type: ignore[method-assign]  # FastAPI's public custom OpenAPI hook.
         return application
 
+    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", {})
+        security_schemes = components.setdefault("securitySchemes", {})
+        self._validate_security_scheme_conflicts(security_schemes, expected_schemes)
+        security_schemes.update(
+            {name: scheme for name, scheme in expected_schemes.items() if name not in security_schemes}
+        )
+
+    def _document_operations(self, openapi_schema: dict[str, typing.Any]) -> None:
+        for path, path_item in openapi_schema.get("paths", {}).items():
+            if path.startswith("x-"):
+                continue
+            for method, operation in path_item.items():
+                if method not in SUPPORTED_HTTP_METHODS:
+                    continue
+                documentation = self._build_version_documentation(
+                    path,
+                    method,
+                    operation.get("description"),
+                    operation.get("x-accept-versioning"),
+                    has_existing_extension="x-accept-versioning" in operation,
+                )
+                if documentation is not None:
+                    operation["x-accept-versioning"], operation["description"] = documentation
+
 
 @FastApiBootstrapper.use_instrument()
 class FastApiCorsInstrument(CorsInstrument):

microbootstrap/bootstrappers/faststream.py

@@ -57,7 +57,11 @@ def _isolate_faststream_subscribers(application: AsgiFastStream) -> None:
 class KwargsAsgiFastStream(AsgiFastStream):
     def __init__(self, **kwargs: typing.Any) -> None:  # noqa: ANN401
         # `broker` argument is positional-only
-        super().__init__(kwargs.pop("broker", None), **kwargs)
+        broker = kwargs.pop("broker", None)
+        if broker is None:
+            super().__init__(**kwargs)
+        else:
+            super().__init__(broker, **kwargs)
 
 
 class FastStreamBootstrapper(ApplicationBootstrapper[FastStreamSettings, AsgiFastStream, FastStreamConfig]):

microbootstrap/bootstrappers/litestar.py

@@ -1,4 +1,5 @@
 from __future__ import annotations
+import dataclasses
 import typing
 
 import litestar
@@ -25,6 +26,15 @@
     HealthCheckTypedDict,
 )
 from microbootstrap.instruments.logging_instrument import LoggingInstrument
+from microbootstrap.instruments.openapi_security_schemes import (
+    OpenApiApiKeySecurityScheme,
+    OpenApiHttpSecurityScheme,
+    OpenApiOAuth2SecurityScheme,
+    OpenApiOpenIdConnectSecurityScheme,
+    _OpenApiSecurityScheme,
+    serialize_security_schemes,
+)
+from microbootstrap.instruments.openapi_version_docs import SUPPORTED_HTTP_METHODS
 from microbootstrap.instruments.opentelemetry_instrument import OpentelemetryInstrument
 from microbootstrap.instruments.prometheus_instrument import (
     LitestarPrometheusConfig,
@@ -37,6 +47,17 @@
 from microbootstrap.settings import LitestarSettings
 
 
+ApplicationT = typing.TypeVar("ApplicationT", bound=litestar.Litestar)
+
+
+@dataclasses.dataclass
+class AcceptVersionedOperation(openapi.spec.Operation):
+    accept_versioning: dict[str, str | list[str]] | None = dataclasses.field(
+        default=None,
+        metadata={"alias": "x-accept-versioning"},
+    )
+
+
 if typing.TYPE_CHECKING:
     from litestar.contrib.opentelemetry import OpenTelemetryConfig
     from litestar.types import ASGIApp, Scope
@@ -102,6 +123,102 @@ 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
+        ):
+            return application
+        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
+
+    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
+        if security_schemes is not None:
+            canonical_schemes: typing.Final = {
+                name: scheme.to_schema() if isinstance(scheme, openapi.spec.SecurityScheme) else scheme
+                for name, scheme in security_schemes.items()
+            }
+            self._validate_security_scheme_conflicts(canonical_schemes, expected_schemes)
+        configured_schemes = {
+            scheme_name: self._build_litestar_security_scheme(security_scheme)
+            for scheme_name, security_scheme in self.instrument_config.security_schemes.items()
+        }
+        if security_schemes is None:
+            openapi_schema.components.security_schemes = typing.cast(
+                "dict[str, openapi.spec.SecurityScheme | openapi.spec.Reference]",
+                configured_schemes,
+            )
+            return
+        security_schemes.update(
+            {name: scheme for name, scheme in configured_schemes.items() if name not in security_schemes}
+        )
+
+    def _document_operations(self, openapi_schema: openapi.spec.OpenAPI) -> None:
+        if openapi_schema.paths is None:
+            return
+        for path, path_item in openapi_schema.paths.items():
+            for method in SUPPORTED_HTTP_METHODS:
+                operation = getattr(path_item, method)
+                if operation is None:
+                    continue
+                existing_extension = (
+                    operation.accept_versioning if isinstance(operation, AcceptVersionedOperation) else None
+                )
+                documentation = self._build_version_documentation(
+                    path,
+                    method,
+                    operation.description,
+                    existing_extension,
+                    has_existing_extension=existing_extension is not None,
+                )
+                if documentation is None:
+                    continue
+                extension, description = documentation
+                if type(operation) is openapi.spec.Operation:
+                    init_fields = {
+                        field.name: getattr(operation, field.name)
+                        for field in dataclasses.fields(openapi.spec.Operation)
+                        if field.init
+                    }
+                    operation = AcceptVersionedOperation(**init_fields, accept_versioning=extension)
+                    setattr(path_item, method, operation)
+                elif type(operation) is not AcceptVersionedOperation:
+                    message = (
+                        f"OpenAPI operation {type(operation).__name__} is not supported "
+                        "for Accept version documentation."
+                    )
+                    raise TypeError(message)
+                operation.accept_versioning = extension
+                operation.description = description
+
+    @classmethod
+    def _build_litestar_security_scheme(cls, security_scheme: _OpenApiSecurityScheme) -> openapi.spec.SecurityScheme:
+        if not isinstance(
+            security_scheme,
+            (
+                OpenApiHttpSecurityScheme,
+                OpenApiApiKeySecurityScheme,
+                OpenApiOAuth2SecurityScheme,
+                OpenApiOpenIdConnectSecurityScheme,
+            ),
+        ):
+            raise AssertionError("Unsupported OpenAPI security scheme.")  # noqa: TRY004
+        scheme_data = security_scheme.model_dump(by_alias=False, exclude_none=True)
+        if "location" in scheme_data:
+            scheme_data["security_scheme_in"] = scheme_data.pop("location")
+        if "flows" in scheme_data:
+            flow_data = scheme_data["flows"]
+            if "resource_owner" in flow_data:
+                flow_data["password"] = flow_data.pop("resource_owner")
+            scheme_data["flows"] = openapi.spec.OAuthFlows(
+                **{flow_name: openapi.spec.OAuthFlow(**flow) for flow_name, flow in flow_data.items()}
+            )
+        return openapi.spec.SecurityScheme(**scheme_data)
+
 
 @LitestarBootstrapper.use_instrument()
 class LitestarCorsInstrument(CorsInstrument):

microbootstrap/helpers.py

@@ -26,7 +26,7 @@ def dataclass_to_dict_no_defaults(dataclass_to_convert: "_DataclassT") -> dict[s
         if dataclass_field.default != value and isinstance(dataclass_field.default_factory, _MISSING_TYPE):
             conversion_result[dataclass_field.name] = value
             continue
-        if value != dataclass_field.default and value != dataclass_field.default_factory():  # type: ignore[misc]
+        if value != dataclass_field.default and value != dataclass_field.default_factory():  # type: ignore[operator]
             conversion_result[dataclass_field.name] = value
 
     return conversion_result