move healthchecks inside library

microbootstrap/bootstrappers/fastapi.py

@@ -10,7 +10,7 @@
 from microbootstrap.bootstrappers.base import ApplicationBootstrapper
 from microbootstrap.config.fastapi import FastApiConfig
 from microbootstrap.instruments.cors_instrument import CorsInstrument
-from microbootstrap.instruments.health_checks_instrument import HealthChecksInstrument
+from microbootstrap.instruments.health_checks_instrument import HealthChecksInstrument, HealthCheckTypedDict
 from microbootstrap.instruments.logging_instrument import LoggingInstrument
 from microbootstrap.instruments.opentelemetry_instrument import OpentelemetryInstrument
 from microbootstrap.instruments.prometheus_instrument import FastApiPrometheusConfig, PrometheusInstrument
@@ -20,10 +20,6 @@
 from microbootstrap.settings import FastApiSettings
 
 
-with contextlib.suppress(ImportError):
-    from health_checks.fastapi_healthcheck import build_fastapi_health_check_router
-
-
 ApplicationT = typing.TypeVar("ApplicationT", bound=fastapi.FastAPI)
 
 
@@ -131,12 +127,18 @@ def get_config_type(cls) -> type[FastApiPrometheusConfig]:
 
 @FastApiBootstrapper.use_instrument()
 class FastApiHealthChecksInstrument(HealthChecksInstrument):
-    def bootstrap_after(self, application: ApplicationT) -> ApplicationT:
-        application.include_router(
-            build_fastapi_health_check_router(
-                health_check=self.health_check,
-                health_check_endpoint=self.instrument_config.health_checks_path,
-                include_in_schema=self.instrument_config.health_checks_include_in_schema,
-            ),
+    def build_fastapi_health_check_router(self) -> fastapi.APIRouter:
+        fastapi_router: typing.Final = fastapi.APIRouter(
+            tags=["probes"],
+            include_in_schema=self.instrument_config.health_checks_include_in_schema,
         )
+
+        @fastapi_router.get(self.instrument_config.health_checks_path)
+        async def health_check_handler() -> HealthCheckTypedDict:
+            return self.render_health_check_data()
+
+        return fastapi_router
+
+    def bootstrap_after(self, application: ApplicationT) -> ApplicationT:
+        application.include_router(self.build_fastapi_health_check_router())
         return application

microbootstrap/bootstrappers/faststream.py

@@ -1,6 +1,4 @@
 from __future__ import annotations
-import contextlib
-import dataclasses
 import json
 import typing
 
@@ -23,10 +21,6 @@
 from microbootstrap.settings import FastStreamSettings
 
 
-with contextlib.suppress(ImportError):
-    from health_checks.http_based import BaseHTTPHealthCheck
-
-
 class KwargsAsgiFastStream(AsgiFastStream):
     def __init__(self, **kwargs: typing.Any) -> None:  # noqa: ANN401
         # `broker` argument is positional-only
@@ -102,33 +96,25 @@ def get_config_type(cls) -> type[FastStreamPrometheusConfig]:
         return FastStreamPrometheusConfig
 
 
-@dataclasses.dataclass
-class FastStreamHealthCheck(BaseHTTPHealthCheck):
-    application: AsgiFastStream | None = None
-
-    async def update_health_status(self) -> bool:
-        return await self.application.broker.ping(timeout=5) if self.application and self.application.broker else False
-
-
 @FastStreamBootstrapper.use_instrument()
 class FastStreamHealthChecksInstrument(HealthChecksInstrument):
     def bootstrap(self) -> None: ...
     def bootstrap_before(self) -> dict[str, typing.Any]:
         @handle_get
         async def check_health(scope: typing.Any) -> AsgiResponse:  # noqa: ANN401, ARG001
-            health_check_data = await self.health_check.check_health()
             return (
-                AsgiResponse(json.dumps(health_check_data).encode(), 200, headers={"content-type": "text/plain"})
-                if health_check_data["health_status"]
+                AsgiResponse(
+                    json.dumps(self.render_health_check_data()).encode(), 200, headers={"content-type": "text/plain"}
+                )
+                if await self.define_health_status()
                 else AsgiResponse(b"Service is unhealthy", 500, headers={"content-type": "application/json"})
             )
 
         return {"asgi_routes": ((self.instrument_config.health_checks_path, check_health),)}
 
+    async def define_health_status(self) -> bool:
+        return await self.application.broker.ping(timeout=5) if self.application and self.application.broker else False
+
     def bootstrap_after(self, application: AsgiFastStream) -> AsgiFastStream:  # type: ignore[override]
-        self.health_check = FastStreamHealthCheck(
-            service_version=self.instrument_config.service_version,
-            service_name=self.instrument_config.service_name,
-            application=application,
-        )
+        self.application = application
         return application

microbootstrap/bootstrappers/litestar.py

@@ -1,8 +1,8 @@
 from __future__ import annotations
-import contextlib
 import typing
 
 import litestar
+import litestar.exceptions
 import litestar.types
 import typing_extensions
 from litestar import openapi
@@ -16,7 +16,7 @@
 from microbootstrap.bootstrappers.base import ApplicationBootstrapper
 from microbootstrap.config.litestar import LitestarConfig
 from microbootstrap.instruments.cors_instrument import CorsInstrument
-from microbootstrap.instruments.health_checks_instrument import HealthChecksInstrument
+from microbootstrap.instruments.health_checks_instrument import HealthChecksInstrument, HealthCheckTypedDict
 from microbootstrap.instruments.logging_instrument import LoggingInstrument
 from microbootstrap.instruments.opentelemetry_instrument import OpentelemetryInstrument
 from microbootstrap.instruments.prometheus_instrument import LitestarPrometheusConfig, PrometheusInstrument
@@ -26,10 +26,6 @@
 from microbootstrap.settings import LitestarSettings
 
 
-with contextlib.suppress(ImportError):
-    from health_checks.litestar_healthcheck import build_litestar_health_check_router
-
-
 class LitestarBootstrapper(
     ApplicationBootstrapper[LitestarSettings, litestar.Litestar, LitestarConfig],
 ):
@@ -150,13 +146,17 @@ def get_config_type(cls) -> type[LitestarPrometheusConfig]:
 
 @LitestarBootstrapper.use_instrument()
 class LitestarHealthChecksInstrument(HealthChecksInstrument):
+    def build_litestar_health_check_router(self) -> litestar.Router:
+        @litestar.get(media_type=litestar.MediaType.JSON)
+        async def health_check_handler() -> HealthCheckTypedDict:
+            return self.render_health_check_data()
+
+        return litestar.Router(
+            path=self.instrument_config.health_checks_path,
+            route_handlers=[health_check_handler],
+            tags=["probes"],
+            include_in_schema=self.instrument_config.health_checks_include_in_schema,
+        )
+
     def bootstrap_before(self) -> dict[str, typing.Any]:
-        return {
-            "route_handlers": [
-                build_litestar_health_check_router(
-                    health_check=self.health_check,
-                    health_check_endpoint=self.instrument_config.health_checks_path,
-                    include_in_schema=self.instrument_config.health_checks_include_in_schema,
-                ),
-            ],
-        }
+        return {"route_handlers": [self.build_litestar_health_check_router()]}

microbootstrap/granian_server.py

@@ -26,8 +26,8 @@ def create_granian_server(
     target: str,
     settings: ServerConfig,
     **granian_options: typing.Any,  # noqa: ANN401
-) -> granian.Granian:
-    return granian.Granian(
+) -> granian.Granian:  # type: ignore[name-defined]
+    return granian.Granian(  # type: ignore[attr-defined]
         target=target,
         address=settings.server_host,
         port=settings.server_port,

microbootstrap/instruments/health_checks_instrument.py

@@ -1,16 +1,15 @@
 from __future__ import annotations
-import contextlib
 import typing
 
-from microbootstrap.instruments.base import BaseInstrumentConfig, Instrument
-
+import typing_extensions
 
-if typing.TYPE_CHECKING:
-    from health_checks.base import HealthCheck
+from microbootstrap.instruments.base import BaseInstrumentConfig, Instrument
 
 
-with contextlib.suppress(ImportError):
-    from health_checks.http_based import DefaultHTTPHealthCheck
+class HealthCheckTypedDict(typing_extensions.TypedDict, total=False):
+    service_version: typing.Optional[str]  # noqa: UP007 (Litestar fails to build OpenAPI schema on Python 3.9)
+    service_name: typing.Optional[str]  # noqa: UP007 (Litestar fails to build OpenAPI schema on Python 3.9)
+    health_status: bool
 
 
 class HealthChecksConfig(BaseInstrumentConfig):
@@ -26,11 +25,12 @@ class HealthChecksInstrument(Instrument[HealthChecksConfig]):
     instrument_name = "Health checks"
     ready_condition = "Set health_checks_enabled to True"
 
-    def bootstrap(self) -> None:
-        self.health_check: HealthCheck = DefaultHTTPHealthCheck(
-            service_version=self.instrument_config.service_version,
-            service_name=self.instrument_config.service_name,
-        )
+    def render_health_check_data(self) -> HealthCheckTypedDict:
+        return {
+            "service_version": self.instrument_config.service_version,
+            "service_name": self.instrument_config.service_name,
+            "health_status": True,
+        }
 
     def is_ready(self) -> bool:
         return self.instrument_config.health_checks_enabled

pyproject.toml

@@ -49,15 +49,13 @@ authors = [{ name = "community-of-python" }]
 fastapi = [
     "fastapi>=0.100",
     "fastapi-offline-docs>=1",
-    "health-checks>=1",
     "opentelemetry-instrumentation-asgi>=0.46b0",
     "opentelemetry-instrumentation-fastapi>=0.46b0",
     "prometheus-fastapi-instrumentator>=6.1",
 ]
 litestar = [
     "litestar>=2.9",
     "litestar-offline-docs>=1",
-    "health-checks>=1",
     "opentelemetry-instrumentation-asgi>=0.46b0",
     "prometheus-client>=0.20",
 ]

tests/test_granian_server.py

@@ -5,4 +5,4 @@
 
 
 def test_granian_server(minimal_server_config: ServerConfig) -> None:
-    assert isinstance(create_granian_server("some:app", minimal_server_config), granian.Granian)
+    assert isinstance(create_granian_server("some:app", minimal_server_config), granian.Granian)  # type: ignore[attr-defined]