diff --git a/lite_bootstrap/bootstrappers/fastapi_bootstrapper.py b/lite_bootstrap/bootstrappers/fastapi_bootstrapper.py index a307b6d..8f68348 100644 --- a/lite_bootstrap/bootstrappers/fastapi_bootstrapper.py +++ b/lite_bootstrap/bootstrappers/fastapi_bootstrapper.py @@ -155,9 +155,7 @@ async def send_wrapper(message: "Message") -> None: @dataclasses.dataclass(kw_only=True) -class FastAPILoggingInstrument(LoggingInstrument): - bootstrap_config: FastAPIConfig # ty: ignore[invalid-mutable-override] - +class FastAPILoggingInstrument(LoggingInstrument[FastAPIConfig]): def bootstrap(self) -> None: super().bootstrap() if not self.bootstrap_config.fastapi_logging_middleware_enabled: @@ -166,17 +164,13 @@ def bootstrap(self) -> None: @dataclasses.dataclass(kw_only=True, slots=True) -class FastAPICorsInstrument(CorsInstrument): - bootstrap_config: FastAPIConfig # ty: ignore[invalid-mutable-override] - +class FastAPICorsInstrument(CorsInstrument[FastAPIConfig]): def bootstrap(self) -> None: self.bootstrap_config.app.add_middleware(CORSMiddleware, **self.cors_kwargs) @dataclasses.dataclass(kw_only=True, slots=True) -class FastAPIHealthChecksInstrument(HealthChecksInstrument): - bootstrap_config: FastAPIConfig # ty: ignore[invalid-mutable-override] - +class FastAPIHealthChecksInstrument(HealthChecksInstrument[FastAPIConfig]): def build_fastapi_health_check_router(self) -> "fastapi.APIRouter": fastapi_router = fastapi.APIRouter( tags=["probes"], @@ -194,9 +188,7 @@ def bootstrap(self) -> None: @dataclasses.dataclass(kw_only=True) -class FastAPIOpenTelemetryInstrument(OpenTelemetryInstrument): - bootstrap_config: FastAPIConfig # ty: ignore[invalid-mutable-override] - +class FastAPIOpenTelemetryInstrument(OpenTelemetryInstrument[FastAPIConfig]): def bootstrap(self) -> None: super().bootstrap() FastAPIInstrumentor.instrument_app( @@ -213,8 +205,7 @@ def teardown(self) -> None: @dataclasses.dataclass(kw_only=True) -class FastAPIPrometheusInstrument(PrometheusInstrument): - bootstrap_config: FastAPIConfig # ty: ignore[invalid-mutable-override] +class FastAPIPrometheusInstrument(PrometheusInstrument[FastAPIConfig]): missing_dependency_message = "prometheus_fastapi_instrumentator is not installed" @staticmethod @@ -236,9 +227,7 @@ def bootstrap(self) -> None: @dataclasses.dataclass(kw_only=True) -class FastAPISwaggerInstrument(SwaggerInstrument): - bootstrap_config: FastAPIConfig # ty: ignore[invalid-mutable-override] - +class FastAPISwaggerInstrument(SwaggerInstrument[FastAPIConfig]): def bootstrap(self) -> None: config = self.bootstrap_config application = config.app diff --git a/lite_bootstrap/bootstrappers/fastmcp_bootstrapper.py b/lite_bootstrap/bootstrappers/fastmcp_bootstrapper.py index 4851f2a..9c2dae4 100644 --- a/lite_bootstrap/bootstrappers/fastmcp_bootstrapper.py +++ b/lite_bootstrap/bootstrappers/fastmcp_bootstrapper.py @@ -199,9 +199,7 @@ class FastMcpConfig( @dataclasses.dataclass(kw_only=True) -class FastMcpHealthChecksInstrument(HealthChecksInstrument): - bootstrap_config: FastMcpConfig # ty: ignore[invalid-mutable-override] - +class FastMcpHealthChecksInstrument(HealthChecksInstrument[FastMcpConfig]): def bootstrap(self) -> None: config = self.bootstrap_config @@ -216,8 +214,7 @@ async def health_check_handler(_: "Request") -> "JSONResponse": @dataclasses.dataclass(kw_only=True) -class FastMcpOpenTelemetryInstrument(OpenTelemetryInstrument): - bootstrap_config: FastMcpConfig # ty: ignore[invalid-mutable-override] +class FastMcpOpenTelemetryInstrument(OpenTelemetryInstrument[FastMcpConfig]): missing_dependency_message = "opentelemetry-instrumentation-asgi is not installed" _restore_http_app: typing.Callable[[], None] | None = dataclasses.field( default=None, init=False, repr=False, compare=False @@ -257,8 +254,7 @@ def teardown(self) -> None: @dataclasses.dataclass(kw_only=True) -class FastMcpPrometheusInstrument(PrometheusInstrument): - bootstrap_config: FastMcpConfig # ty: ignore[invalid-mutable-override] +class FastMcpPrometheusInstrument(PrometheusInstrument[FastMcpConfig]): missing_dependency_message = "prometheus_client or prometheus_fastapi_instrumentator is not installed" _restore_http_app: typing.Callable[[], None] | None = dataclasses.field( default=None, init=False, repr=False, compare=False @@ -319,9 +315,7 @@ def teardown(self) -> None: @dataclasses.dataclass(kw_only=True) -class FastMcpLoggingInstrument(LoggingInstrument): - bootstrap_config: FastMcpConfig # ty: ignore[invalid-mutable-override] - +class FastMcpLoggingInstrument(LoggingInstrument[FastMcpConfig]): def bootstrap(self) -> None: super().bootstrap() if not self.bootstrap_config.fastmcp_logging_middleware_enabled: diff --git a/lite_bootstrap/bootstrappers/faststream_bootstrapper.py b/lite_bootstrap/bootstrappers/faststream_bootstrapper.py index 1b7b167..4e4dfa9 100644 --- a/lite_bootstrap/bootstrappers/faststream_bootstrapper.py +++ b/lite_bootstrap/bootstrappers/faststream_bootstrapper.py @@ -81,9 +81,7 @@ class FastStreamConfig( @dataclasses.dataclass(kw_only=True, slots=True) -class FastStreamHealthChecksInstrument(HealthChecksInstrument): - bootstrap_config: FastStreamConfig # ty: ignore[invalid-mutable-override] - +class FastStreamHealthChecksInstrument(HealthChecksInstrument[FastStreamConfig]): def bootstrap(self) -> None: config = self.bootstrap_config @@ -113,8 +111,7 @@ async def _define_health_status(self) -> bool: @dataclasses.dataclass(kw_only=True) -class FastStreamLoggingInstrument(LoggingInstrument): - bootstrap_config: FastStreamConfig # ty: ignore[invalid-mutable-override] +class FastStreamLoggingInstrument(LoggingInstrument[FastStreamConfig]): _prior_broker_params_storage: typing.Any = dataclasses.field(default=None, init=False, repr=False, compare=False) _broker_logger_replaced: bool = dataclasses.field(default=False, init=False, repr=False, compare=False) @@ -141,9 +138,7 @@ def teardown(self) -> None: @dataclasses.dataclass(kw_only=True) -class FastStreamOpenTelemetryInstrument(OpenTelemetryInstrument): - bootstrap_config: FastStreamConfig # ty: ignore[invalid-mutable-override] - +class FastStreamOpenTelemetryInstrument(OpenTelemetryInstrument[FastStreamConfig]): def bootstrap(self) -> None: super().bootstrap() config = self.bootstrap_config @@ -160,8 +155,7 @@ def _make_collector_registry() -> "prometheus_client.CollectorRegistry": @dataclasses.dataclass(kw_only=True) -class FastStreamPrometheusInstrument(PrometheusInstrument): - bootstrap_config: FastStreamConfig # ty: ignore[invalid-mutable-override] +class FastStreamPrometheusInstrument(PrometheusInstrument[FastStreamConfig]): collector_registry: "prometheus_client.CollectorRegistry" = dataclasses.field(init=False) not_configured_reason = ( PrometheusInstrument.not_configured_reason diff --git a/lite_bootstrap/bootstrappers/litestar_bootstrapper.py b/lite_bootstrap/bootstrappers/litestar_bootstrapper.py index 83d7293..c2c73d0 100644 --- a/lite_bootstrap/bootstrappers/litestar_bootstrapper.py +++ b/lite_bootstrap/bootstrappers/litestar_bootstrapper.py @@ -139,17 +139,13 @@ def __post_init__(self) -> None: @dataclasses.dataclass(kw_only=True, slots=True) -class LitestarCorsInstrument(CorsInstrument): - bootstrap_config: LitestarConfig # ty: ignore[invalid-mutable-override] - +class LitestarCorsInstrument(CorsInstrument[LitestarConfig]): def bootstrap(self) -> None: self.bootstrap_config.application_config.cors_config = CORSConfig(**self.cors_kwargs) @dataclasses.dataclass(kw_only=True, slots=True) -class LitestarHealthChecksInstrument(HealthChecksInstrument): - bootstrap_config: LitestarConfig # ty: ignore[invalid-mutable-override] - +class LitestarHealthChecksInstrument(HealthChecksInstrument[LitestarConfig]): def build_litestar_health_check_router(self) -> "litestar.Router": @litestar.get(media_type=litestar.MediaType.JSON) async def health_check_handler() -> HealthCheckTypedDict: @@ -167,9 +163,7 @@ def bootstrap(self) -> None: @dataclasses.dataclass(kw_only=True) -class LitestarLoggingInstrument(LoggingInstrument): - bootstrap_config: LitestarConfig # ty: ignore[invalid-mutable-override] - +class LitestarLoggingInstrument(LoggingInstrument[LitestarConfig]): def _build_logging_middleware_excluded_paths(self) -> list[str]: """Regex-escaped path prefixes for infrastructure routes not worth an access log line.""" # Litestar matches exclude patterns with an unanchored search, so anchor each one to the @@ -209,8 +203,7 @@ def _configure_structlog_loggers(self) -> None: @dataclasses.dataclass(kw_only=True) -class LitestarOpenTelemetryInstrument(OpenTelemetryInstrument): - bootstrap_config: LitestarConfig # ty: ignore[invalid-mutable-override] +class LitestarOpenTelemetryInstrument(OpenTelemetryInstrument[LitestarConfig]): missing_dependency_message = "opentelemetry-instrumentation-asgi or litestar>=2.22 is not installed" @staticmethod @@ -251,8 +244,7 @@ def bootstrap(self) -> None: @dataclasses.dataclass(kw_only=True) -class LitestarPrometheusInstrument(PrometheusInstrument): - bootstrap_config: LitestarConfig # ty: ignore[invalid-mutable-override] +class LitestarPrometheusInstrument(PrometheusInstrument[LitestarConfig]): missing_dependency_message = "prometheus_client is not installed" @staticmethod @@ -279,12 +271,11 @@ class LitestarPrometheusController(PrometheusController): @dataclasses.dataclass(kw_only=True) -class LitestarSwaggerInstrument(SwaggerInstrument): - bootstrap_config: LitestarConfig # ty: ignore[invalid-mutable-override] +class LitestarSwaggerInstrument(SwaggerInstrument[LitestarConfig]): not_configured_reason = "swagger_path is empty or not valid" @classmethod - def is_configured(cls, bootstrap_config: "LitestarConfig") -> bool: # ty: ignore[invalid-method-override] + def is_configured(cls, bootstrap_config: "LitestarConfig") -> bool: return bool(bootstrap_config.swagger_path) and is_valid_path(bootstrap_config.swagger_path) def bootstrap(self) -> None: diff --git a/lite_bootstrap/instruments/cors_instrument.py b/lite_bootstrap/instruments/cors_instrument.py index 16913df..d97a978 100644 --- a/lite_bootstrap/instruments/cors_instrument.py +++ b/lite_bootstrap/instruments/cors_instrument.py @@ -32,8 +32,11 @@ def __post_init__(self) -> None: super().__post_init__() +CorsConfigT = typing.TypeVar("CorsConfigT", bound=CorsConfig) + + @dataclasses.dataclass(kw_only=True, slots=True) -class CorsInstrument(BaseInstrument[CorsConfig]): +class CorsInstrument(BaseInstrument[CorsConfigT]): not_configured_reason = "cors_allowed_origins or cors_allowed_origin_regex must be provided" @classmethod diff --git a/lite_bootstrap/instruments/healthchecks_instrument.py b/lite_bootstrap/instruments/healthchecks_instrument.py index 71647ba..035b91c 100644 --- a/lite_bootstrap/instruments/healthchecks_instrument.py +++ b/lite_bootstrap/instruments/healthchecks_instrument.py @@ -1,4 +1,5 @@ import dataclasses +import typing import typing_extensions @@ -26,8 +27,11 @@ def health_check_data(self) -> HealthCheckTypedDict: } +HealthChecksConfigT = typing.TypeVar("HealthChecksConfigT", bound=HealthChecksConfig) + + @dataclasses.dataclass(kw_only=True, slots=True) -class HealthChecksInstrument(BaseInstrument[HealthChecksConfig]): +class HealthChecksInstrument(BaseInstrument[HealthChecksConfigT]): not_configured_reason = "health_checks_enabled is False" @classmethod diff --git a/lite_bootstrap/instruments/logging_instrument.py b/lite_bootstrap/instruments/logging_instrument.py index be8b9c4..24c3b1d 100644 --- a/lite_bootstrap/instruments/logging_instrument.py +++ b/lite_bootstrap/instruments/logging_instrument.py @@ -78,8 +78,11 @@ class LoggingConfig(BaseConfig): logging_enabled: bool = True +LoggingConfigT = typing.TypeVar("LoggingConfigT", bound=LoggingConfig) + + @dataclasses.dataclass(kw_only=True, slots=True) -class LoggingInstrument(BaseInstrument[LoggingConfig]): +class LoggingInstrument(BaseInstrument[LoggingConfigT]): not_configured_reason = "logging_enabled is False" missing_dependency_message = "structlog is not installed" _logger_factory: "MemoryLoggerFactory | None" = dataclasses.field( diff --git a/lite_bootstrap/instruments/opentelemetry_instrument.py b/lite_bootstrap/instruments/opentelemetry_instrument.py index 9da3961..98cd858 100644 --- a/lite_bootstrap/instruments/opentelemetry_instrument.py +++ b/lite_bootstrap/instruments/opentelemetry_instrument.py @@ -145,8 +145,11 @@ def force_flush(self, timeout_millis: int = 30000) -> bool: # noqa: ARG002 return True # pragma: no cover - matters only at the sdk 1.28 floor, whose base returns None +OpenTelemetryConfigT = typing.TypeVar("OpenTelemetryConfigT", bound=OpenTelemetryConfig) + + @dataclasses.dataclass(kw_only=True, slots=True) -class OpenTelemetryInstrument(BaseInstrument[OpenTelemetryConfig]): +class OpenTelemetryInstrument(BaseInstrument[OpenTelemetryConfigT]): """OpenTelemetry tracing instrument. Lifecycle note: ``bootstrap()`` calls ``opentelemetry.trace.set_tracer_provider``, diff --git a/lite_bootstrap/instruments/prometheus_instrument.py b/lite_bootstrap/instruments/prometheus_instrument.py index 694f918..def18bb 100644 --- a/lite_bootstrap/instruments/prometheus_instrument.py +++ b/lite_bootstrap/instruments/prometheus_instrument.py @@ -1,6 +1,7 @@ """Prometheus config and readiness check; framework-specific bootstrap lives in the bootstrapper subclasses.""" import dataclasses +import typing from lite_bootstrap.helpers.path import is_valid_path from lite_bootstrap.instruments.base import BaseConfig, BaseInstrument @@ -12,8 +13,11 @@ class PrometheusConfig(BaseConfig): prometheus_metrics_include_in_schema: bool = False +PrometheusConfigT = typing.TypeVar("PrometheusConfigT", bound=PrometheusConfig) + + @dataclasses.dataclass(kw_only=True, slots=True) -class PrometheusInstrument(BaseInstrument[PrometheusConfig]): +class PrometheusInstrument(BaseInstrument[PrometheusConfigT]): not_configured_reason = "prometheus_metrics_path is empty or not valid" @classmethod diff --git a/lite_bootstrap/instruments/swagger_instrument.py b/lite_bootstrap/instruments/swagger_instrument.py index 91eea86..8ee8ac0 100644 --- a/lite_bootstrap/instruments/swagger_instrument.py +++ b/lite_bootstrap/instruments/swagger_instrument.py @@ -1,6 +1,7 @@ """Swagger config and minimal base instrument; framework-specific behavior lives in the bootstrapper subclasses.""" import dataclasses +import typing from lite_bootstrap.instruments.base import BaseConfig, BaseInstrument @@ -12,6 +13,9 @@ class SwaggerConfig(BaseConfig): swagger_offline_docs: bool = False +SwaggerConfigT = typing.TypeVar("SwaggerConfigT", bound=SwaggerConfig) + + @dataclasses.dataclass(kw_only=True, slots=True) -class SwaggerInstrument(BaseInstrument[SwaggerConfig]): +class SwaggerInstrument(BaseInstrument[SwaggerConfigT]): pass