merge configs

lite_bootstrap/__init__.py

@@ -1,44 +1,13 @@
-from lite_bootstrap.bootstrappers.fastapi_bootstrapper import (
-    FastAPIBootstrapper,
-    FastAPIHealthChecksInstrument,
-    FastAPILoggingInstrument,
-    FastAPIOpenTelemetryInstrument,
-    FastAPIPrometheusInstrument,
-    FastAPISentryInstrument,
-)
-from lite_bootstrap.bootstrappers.free_bootstrapper import FreeBootstrapper
-from lite_bootstrap.bootstrappers.litestar_bootstrapper import (
-    LitestarBootstrapper,
-    LitestarHealthChecksInstrument,
-    LitestarLoggingInstrument,
-    LitestarOpenTelemetryInstrument,
-    LitestarPrometheusInstrument,
-    LitestarSentryInstrument,
-)
-from lite_bootstrap.instruments.healthchecks_instrument import HealthChecksInstrument
-from lite_bootstrap.instruments.logging_instrument import LoggingInstrument
-from lite_bootstrap.instruments.opentelemetry_instrument import OpenTelemetryInstrument
-from lite_bootstrap.instruments.sentry_instrument import SentryInstrument
-from lite_bootstrap.service_config import ServiceConfig
+from lite_bootstrap.bootstrappers.fastapi_bootstrapper import FastAPIBootstrapper, FastAPIConfig
+from lite_bootstrap.bootstrappers.free_bootstrapper import FreeBootstrapper, FreeBootstrapperConfig
+from lite_bootstrap.bootstrappers.litestar_bootstrapper import LitestarBootstrapper, LitestarConfig
 
 
 __all__ = [
     "FastAPIBootstrapper",
-    "FastAPIHealthChecksInstrument",
-    "FastAPILoggingInstrument",
-    "FastAPIOpenTelemetryInstrument",
-    "FastAPIPrometheusInstrument",
-    "FastAPISentryInstrument",
+    "FastAPIConfig",
     "FreeBootstrapper",
-    "HealthChecksInstrument",
+    "FreeBootstrapperConfig",
     "LitestarBootstrapper",
-    "LitestarHealthChecksInstrument",
-    "LitestarLoggingInstrument",
-    "LitestarOpenTelemetryInstrument",
-    "LitestarPrometheusInstrument",
-    "LitestarSentryInstrument",
-    "LoggingInstrument",
-    "OpenTelemetryInstrument",
-    "SentryInstrument",
-    "ServiceConfig",
+    "LitestarConfig",
 ]

lite_bootstrap/bootstrappers/base.py

@@ -1,26 +1,34 @@
 import abc
 import typing
 
-from lite_bootstrap.instruments.base import BaseInstrument
-from lite_bootstrap.service_config import ServiceConfig
-from lite_bootstrap.types import ApplicationT, BootstrapObjectT
+from lite_bootstrap.instruments.base import BaseConfig, BaseInstrument
+from lite_bootstrap.types import ApplicationT
 
 
-class BaseBootstrapper(abc.ABC, typing.Generic[BootstrapObjectT, ApplicationT]):
-    bootstrap_object: BootstrapObjectT
-    instruments: typing.Sequence[BaseInstrument]
-    service_config: ServiceConfig
+InstrumentT = typing.TypeVar("InstrumentT", bound=BaseInstrument)
+
+
+class BaseBootstrapper(abc.ABC, typing.Generic[ApplicationT]):
+    instruments_types: typing.ClassVar[list[type[BaseInstrument]]]
+    instruments: list[BaseInstrument]
+    bootstrap_config: BaseConfig
+
+    def __init__(self, bootstrap_config: BaseConfig) -> None:
+        self.bootstrap_config = bootstrap_config
+        self.instruments = []
+        for instrument_type in self.instruments_types:
+            instrument = instrument_type(bootstrap_config=bootstrap_config)
+            if instrument.is_ready():
+                self.instruments.append(instrument)
 
     @abc.abstractmethod
     def _prepare_application(self) -> ApplicationT: ...
 
     def bootstrap(self) -> ApplicationT:
         for one_instrument in self.instruments:
-            if one_instrument.is_ready(self.service_config):
-                one_instrument.bootstrap(self.service_config, self.bootstrap_object)
+            one_instrument.bootstrap()
         return self._prepare_application()
 
     def teardown(self) -> None:
         for one_instrument in self.instruments:
-            if one_instrument.is_ready(self.service_config):
-                one_instrument.teardown(self.bootstrap_object)
+            one_instrument.teardown()

lite_bootstrap/bootstrappers/fastapi_bootstrapper.py

@@ -5,12 +5,15 @@
 from prometheus_fastapi_instrumentator import Instrumentator
 
 from lite_bootstrap.bootstrappers.base import BaseBootstrapper
-from lite_bootstrap.instruments.healthchecks_instrument import HealthChecksInstrument, HealthCheckTypedDict
-from lite_bootstrap.instruments.logging_instrument import LoggingInstrument
-from lite_bootstrap.instruments.opentelemetry_instrument import OpenTelemetryInstrument
-from lite_bootstrap.instruments.prometheus_instrument import PrometheusInstrument
-from lite_bootstrap.instruments.sentry_instrument import SentryInstrument
-from lite_bootstrap.service_config import ServiceConfig
+from lite_bootstrap.instruments.healthchecks_instrument import (
+    HealthChecksConfig,
+    HealthChecksInstrument,
+    HealthCheckTypedDict,
+)
+from lite_bootstrap.instruments.logging_instrument import LoggingConfig, LoggingInstrument
+from lite_bootstrap.instruments.opentelemetry_instrument import OpentelemetryConfig, OpenTelemetryInstrument
+from lite_bootstrap.instruments.prometheus_instrument import PrometheusConfig, PrometheusInstrument
+from lite_bootstrap.instruments.sentry_instrument import SentryConfig, SentryInstrument
 
 
 with contextlib.suppress(ImportError):
@@ -19,87 +22,91 @@
     from opentelemetry.trace import get_tracer_provider
 
 
+@dataclasses.dataclass(kw_only=True, slots=True, frozen=True)
+class FastAPIConfig(HealthChecksConfig, LoggingConfig, OpentelemetryConfig, PrometheusConfig, SentryConfig):
+    application: fastapi.FastAPI = dataclasses.field(default_factory=fastapi.FastAPI)
+    opentelemetry_excluded_urls: list[str] = dataclasses.field(default_factory=list)
+    prometheus_instrumentator_params: dict[str, typing.Any] = dataclasses.field(default_factory=dict)
+    prometheus_instrument_params: dict[str, typing.Any] = dataclasses.field(default_factory=dict)
+    prometheus_expose_params: dict[str, typing.Any] = dataclasses.field(default_factory=dict)
+
+
 @dataclasses.dataclass(kw_only=True, slots=True, frozen=True)
 class FastAPIHealthChecksInstrument(HealthChecksInstrument):
-    enabled: bool = True
-    path: str = "/health/"
-    include_in_schema: bool = False
+    bootstrap_config: FastAPIConfig
 
-    def build_fastapi_health_check_router(self, service_config: ServiceConfig) -> fastapi.APIRouter:
+    def build_fastapi_health_check_router(self) -> fastapi.APIRouter:
         fastapi_router = fastapi.APIRouter(
             tags=["probes"],
-            include_in_schema=self.include_in_schema,
+            include_in_schema=self.bootstrap_config.health_checks_include_in_schema,
         )
 
-        @fastapi_router.get(self.path)
+        @fastapi_router.get(self.bootstrap_config.health_checks_path)
         async def health_check_handler() -> HealthCheckTypedDict:
-            return self.render_health_check_data(service_config)
+            return self.render_health_check_data()
 
         return fastapi_router
 
-    def bootstrap(self, service_config: ServiceConfig, application: fastapi.FastAPI | None = None) -> None:
-        if application:
-            application.include_router(self.build_fastapi_health_check_router(service_config))
+    def bootstrap(self) -> None:
+        self.bootstrap_config.application.include_router(self.build_fastapi_health_check_router())
 
 
 @dataclasses.dataclass(kw_only=True, frozen=True)
-class FastAPILoggingInstrument(LoggingInstrument): ...
+class FastAPILoggingInstrument(LoggingInstrument):
+    bootstrap_config: FastAPIConfig
 
 
 @dataclasses.dataclass(kw_only=True, frozen=True)
 class FastAPIOpenTelemetryInstrument(OpenTelemetryInstrument):
-    excluded_urls: list[str] = dataclasses.field(default_factory=list)
+    bootstrap_config: FastAPIConfig
 
-    def bootstrap(self, service_config: ServiceConfig, application: fastapi.FastAPI | None = None) -> None:
-        super().bootstrap(service_config, application)
+    def bootstrap(self) -> None:
+        super().bootstrap()
         FastAPIInstrumentor.instrument_app(
-            app=application,
+            app=self.bootstrap_config.application,
             tracer_provider=get_tracer_provider(),
-            excluded_urls=",".join(self.excluded_urls),
+            excluded_urls=",".join(self.bootstrap_config.opentelemetry_excluded_urls),
         )
 
-    def teardown(self, application: fastapi.FastAPI | None = None) -> None:
-        if application:
-            FastAPIInstrumentor.uninstrument_app(application)
+    def teardown(self) -> None:
+        FastAPIInstrumentor.uninstrument_app(self.bootstrap_config.application)
         super().teardown()
 
 
 @dataclasses.dataclass(kw_only=True, frozen=True)
-class FastAPISentryInstrument(SentryInstrument): ...
+class FastAPISentryInstrument(SentryInstrument):
+    bootstrap_config: FastAPIConfig
 
 
 @dataclasses.dataclass(kw_only=True, frozen=True)
 class FastAPIPrometheusInstrument(PrometheusInstrument):
-    metrics_path: str = "/metrics"
-    metrics_include_in_schema: bool = False
-    instrumentator_params: dict[str, typing.Any] = dataclasses.field(default_factory=dict)
-    instrument_params: dict[str, typing.Any] = dataclasses.field(default_factory=dict)
-    expose_params: dict[str, typing.Any] = dataclasses.field(default_factory=dict)
-
-    def bootstrap(self, _: ServiceConfig, application: fastapi.FastAPI | None = None) -> None:
-        if application:
-            Instrumentator(**self.instrumentator_params).instrument(
-                application,
-                **self.instrument_params,
-            ).expose(
-                application,
-                endpoint=self.metrics_path,
-                include_in_schema=self.metrics_include_in_schema,
-                **self.expose_params,
-            )
+    bootstrap_config: FastAPIConfig
+
+    def bootstrap(self) -> None:
+        Instrumentator(**self.bootstrap_config.prometheus_instrument_params).instrument(
+            self.bootstrap_config.application,
+            **self.bootstrap_config.prometheus_instrument_params,
+        ).expose(
+            self.bootstrap_config.application,
+            endpoint=self.bootstrap_config.prometheus_metrics_path,
+            include_in_schema=self.bootstrap_config.prometheus_metrics_include_in_schema,
+            **self.bootstrap_config.prometheus_expose_params,
+        )
 
 
-@dataclasses.dataclass(kw_only=True, slots=True, frozen=True)
-class FastAPIBootstrapper(BaseBootstrapper[fastapi.FastAPI, fastapi.FastAPI]):
-    bootstrap_object: fastapi.FastAPI
-    instruments: typing.Sequence[
-        FastAPIOpenTelemetryInstrument
-        | FastAPISentryInstrument
-        | FastAPIHealthChecksInstrument
-        | FastAPILoggingInstrument
-        | FastAPIPrometheusInstrument
+class FastAPIBootstrapper(BaseBootstrapper[fastapi.FastAPI]):
+    instruments_types: typing.ClassVar = [
+        FastAPIOpenTelemetryInstrument,
+        FastAPISentryInstrument,
+        FastAPIHealthChecksInstrument,
+        FastAPILoggingInstrument,
+        FastAPIPrometheusInstrument,
     ]
-    service_config: ServiceConfig
+    bootstrap_config: FastAPIConfig
+    __slots__ = "bootstrap_config", "instruments"
+
+    def __init__(self, bootstrap_config: FastAPIConfig) -> None:
+        super().__init__(bootstrap_config)
 
     def _prepare_application(self) -> fastapi.FastAPI:
-        return self.bootstrap_object
+        return self.bootstrap_config.application

lite_bootstrap/bootstrappers/free_bootstrapper.py

@@ -2,17 +2,26 @@
 import typing
 
 from lite_bootstrap.bootstrappers.base import BaseBootstrapper
-from lite_bootstrap.instruments.logging_instrument import LoggingInstrument
-from lite_bootstrap.instruments.opentelemetry_instrument import OpenTelemetryInstrument
-from lite_bootstrap.instruments.sentry_instrument import SentryInstrument
-from lite_bootstrap.service_config import ServiceConfig
+from lite_bootstrap.instruments.logging_instrument import LoggingConfig, LoggingInstrument
+from lite_bootstrap.instruments.opentelemetry_instrument import OpentelemetryConfig, OpenTelemetryInstrument
+from lite_bootstrap.instruments.sentry_instrument import SentryConfig, SentryInstrument
 
 
 @dataclasses.dataclass(kw_only=True, slots=True, frozen=True)
-class FreeBootstrapper(BaseBootstrapper[None, None]):
-    bootstrap_object: None = None
-    instruments: typing.Sequence[OpenTelemetryInstrument | SentryInstrument | LoggingInstrument]
-    service_config: ServiceConfig
+class FreeBootstrapperConfig(LoggingConfig, OpentelemetryConfig, SentryConfig): ...
+
+
+class FreeBootstrapper(BaseBootstrapper[None]):
+    instruments_types: typing.ClassVar = [
+        OpenTelemetryInstrument,
+        SentryInstrument,
+        LoggingInstrument,
+    ]
+    bootstrap_config: FreeBootstrapperConfig
+    __slots__ = "bootstrap_config", "instruments"
+
+    def __init__(self, bootstrap_config: FreeBootstrapperConfig) -> None:
+        super().__init__(bootstrap_config)
 
     def _prepare_application(self) -> None:
-        return self.bootstrap_object
+        return None