Add ASGI FastStream support ✨ (#64)

* Add faststream config

* Update config

* Add FastStream bootstrapper

* Fix annotations

* Update

* Update

* Update

* Add health checks

* Update

* Add prometheus config

* Update

* Update

* Update

* Fix logging

* Update prometheus config

* Update

* Update

* Update

* Update

* Fix tests

* Update

* Update

* Update

Author: vrslev

README.md

@@ -45,8 +45,10 @@ With <b>microbootstrap</b>, you receive an application with lightweight built-in
 
 Those instruments can be bootstrapped for:
 
-- `fastapi`
-- `litestar`
+- `fastapi`,
+- `litestar`,
+- or `faststream` service,
+- or even a service that doesn't use one of these frameworks.
 
 Interested? Let's dive right in ⚡
 
@@ -71,22 +73,30 @@ Interested? Let's dive right in ⚡
 
 ## Installation
 
-You can install the package using either `pip` or `poetry`.
 Also, you can specify extras during installation for concrete framework:
 
 - `fastapi`
 - `litestar`
+- `faststream` (ASGI app)
+
+Also we have `granian` extra that is requires for `create_granian_server`.
+
+For uv:
+
+```bash
+uv add "microbootstrap[fastapi]"
+```
 
 For poetry:
 
 ```bash
-$ poetry add microbootstrap -E fastapi
+poetry add microbootstrap -E fastapi
 ```
 
 For pip:
 
 ```bash
-$ pip install microbootstrap[fastapi]
+pip install "microbootstrap[fastapi]"
 ```
 
 ## Quickstart
@@ -224,7 +234,7 @@ These settings are subsequently passed to the [sentry-sdk](https://pypi.org/proj
 
 ### Prometheus
 
-Prometheus integration presents a challenge because the underlying libraries for `FastAPI` and `Litestar` differ significantly, making it impossible to unify them under a single interface. As a result, the Prometheus settings for `FastAPI` and `Litestar` must be configured separately.
+Prometheus integration presents a challenge because the underlying libraries for `FastAPI`, `Litestar` and `FastStream` differ significantly, making it impossible to unify them under a single interface. As a result, the Prometheus settings for `FastAPI`, `Litestar` and `FastStream` must be configured separately.
 
 #### FastAPI
 
@@ -234,7 +244,7 @@ To bootstrap prometheus you have to provide `prometheus_metrics_path`
 from microbootstrap.settings import FastApiSettings
 
 
-class YourFastApiSettings(FastApiSettings):
+class YourSettings(FastApiSettings):
     service_name: str
 
     prometheus_metrics_path: str = "/metrics"
@@ -255,7 +265,7 @@ Parameters description:
 - `prometheus_instrument_params` - will be passed to `Instrumentor.instrument(...)`.
 - `prometheus_expose_params` - will be passed to `Instrumentor.expose(...)`.
 
-FastApi prometheus bootstrapper uses [prometheus-fastapi-instrumentator](https://github.com/trallnag/prometheus-fastapi-instrumentator) that's why there are three different dict for parameters.
+FastAPI prometheus bootstrapper uses [prometheus-fastapi-instrumentator](https://github.com/trallnag/prometheus-fastapi-instrumentator) that's why there are three different dict for parameters.
 
 #### Litestar
 
@@ -265,7 +275,7 @@ To bootstrap prometheus you have to provide `prometheus_metrics_path`
 from microbootstrap.settings import LitestarSettings
 
 
-class YourFastApiSettings(LitestarSettings):
+class YourSettings(LitestarSettings):
     service_name: str
 
     prometheus_metrics_path: str = "/metrics"
@@ -280,6 +290,30 @@ Parameters description:
 - `prometheus_metrics_path` - path to metrics handler.
 - `prometheus_additional_params` - will be passed to `litestar.contrib.prometheus.PrometheusConfig`.
 
+#### FastStream
+
+To bootstrap prometheus you have to provide `prometheus_metrics_path` and `prometheus_middleware_cls`:
+
+```python
+from microbootstrap import FastStreamSettings
+from faststream.redis.prometheus import RedisPrometheusMiddleware
+
+
+class YourSettings(FastStreamSettings):
+    service_name: str
+
+    prometheus_metrics_path: str = "/metrics"
+    prometheus_middleware_cls: type[FastStreamPrometheusMiddlewareProtocol] | None = RedisPrometheusMiddleware
+
+    ... # Other settings here
+```
+
+Parameters description:
+
+- `service_name` - will be attached to metric's names, there are no name restrictions.
+- `prometheus_metrics_path` - path to metrics handler.
+- `prometheus_middleware_cls` - Prometheus middleware for your broker.
+
 ### Opentelemetry
 
 To bootstrap Opentelemetry, you must provide several parameters:
@@ -293,7 +327,7 @@ To bootstrap Opentelemetry, you must provide several parameters:
 However, additional parameters can also be supplied if needed.
 
 ```python
-from microbootstrap.settings import BaseServiceSettings
+from microbootstrap.settings import BaseServiceSettings, FastStreamPrometheusMiddlewareProtocol
 from microbootstrap.instruments.opentelemetry_instrument import OpenTelemetryInstrumentor
 
 
@@ -326,6 +360,21 @@ Parameters description:
 
 These settings are subsequently passed to [opentelemetry](https://opentelemetry.io/), finalizing your Opentelemetry integration.
 
+#### FastStream
+
+For FastStream you also should pass `opentelemetry_middleware_cls` - OpenTelemetry middleware for your broker
+
+```python
+from microbootstrap import FastStreamSettings, FastStreamTelemetryMiddlewareProtocol
+from faststream.redis.opentelemetry import RedisTelemetryMiddleware
+
+
+class YourSettings(FastStreamSettings):
+    ...
+    opentelemetry_middleware_cls: type[FastStreamTelemetryMiddlewareProtocol] | None = RedisTelemetryMiddleware
+    ...
+```
+
 ### Logging
 
 <b>microbootstrap</b> provides in-memory JSON logging through the use of [structlog](https://pypi.org/project/structlog/).
@@ -412,6 +461,18 @@ 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.
 
+#### FastStream AsyncAPI documentation
+
+AsyncAPI documentation is available by default under `/asyncapi` route. You can change that by setting `asyncapi_path`:
+
+```python
+from microbootstrap import FastStreamSettings
+
+
+class YourSettings(FastStreamSettings):
+    asyncapi_path: str | None = None
+```
+
 ### Health checks
 
 ```python

microbootstrap/__init__.py

@@ -1,17 +1,36 @@
 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.opentelemetry_instrument import OpentelemetryConfig
-from microbootstrap.instruments.prometheus_instrument import FastApiPrometheusConfig, LitestarPrometheusConfig
+from microbootstrap.instruments.opentelemetry_instrument import (
+    FastStreamOpentelemetryConfig,
+    FastStreamTelemetryMiddlewareProtocol,
+    OpentelemetryConfig,
+)
+from microbootstrap.instruments.prometheus_instrument import (
+    FastApiPrometheusConfig,
+    FastStreamPrometheusConfig,
+    FastStreamPrometheusMiddlewareProtocol,
+    LitestarPrometheusConfig,
+)
 from microbootstrap.instruments.sentry_instrument import SentryConfig
 from microbootstrap.instruments.swagger_instrument import SwaggerConfig
-from microbootstrap.settings import FastApiSettings, InstrumentsSetupperSettings, LitestarSettings
+from microbootstrap.settings import (
+    FastApiSettings,
+    FastStreamSettings,
+    InstrumentsSetupperSettings,
+    LitestarSettings,
+)
 
 
 __all__ = (
     "CorsConfig",
     "FastApiPrometheusConfig",
     "FastApiSettings",
+    "FastStreamOpentelemetryConfig",
+    "FastStreamPrometheusConfig",
+    "FastStreamPrometheusMiddlewareProtocol",
+    "FastStreamSettings",
+    "FastStreamTelemetryMiddlewareProtocol",
     "HealthChecksConfig",
     "InstrumentsSetupperSettings",
     "LitestarPrometheusConfig",

microbootstrap/bootstrappers/faststream.py

@@ -0,0 +1,111 @@
+from __future__ import annotations
+import json
+import typing
+
+import prometheus_client
+import structlog
+import typing_extensions
+from faststream.asgi import AsgiFastStream, AsgiResponse
+from faststream.asgi import get as handle_get
+
+from microbootstrap.bootstrappers.base import ApplicationBootstrapper
+from microbootstrap.config.faststream import FastStreamConfig
+from microbootstrap.instruments.health_checks_instrument import HealthChecksInstrument
+from microbootstrap.instruments.logging_instrument import LoggingInstrument
+from microbootstrap.instruments.opentelemetry_instrument import (
+    BaseOpentelemetryInstrument,
+    FastStreamOpentelemetryConfig,
+)
+from microbootstrap.instruments.prometheus_instrument import FastStreamPrometheusConfig, PrometheusInstrument
+from microbootstrap.instruments.sentry_instrument import SentryInstrument
+from microbootstrap.settings import FastStreamSettings
+
+
+class KwargsAsgiFastStream(AsgiFastStream):
+    def __init__(self, **kwargs: typing.Any) -> None:  # noqa: ANN401
+        # `broker` argument is positional-only
+        super().__init__(kwargs.pop("broker", None), **kwargs)
+
+
+class FastStreamBootstrapper(ApplicationBootstrapper[FastStreamSettings, AsgiFastStream, FastStreamConfig]):
+    application_config = FastStreamConfig()
+    application_type = KwargsAsgiFastStream
+
+    def bootstrap_before(self: typing_extensions.Self) -> dict[str, typing.Any]:
+        return {
+            "title": self.settings.service_name,
+            "version": self.settings.service_version,
+            "description": self.settings.service_description,
+            "on_shutdown": [self.teardown],
+            "on_startup": [self.console_writer.print_bootstrap_table],
+            "asyncapi_path": self.settings.asyncapi_path,
+        }
+
+
+FastStreamBootstrapper.use_instrument()(SentryInstrument)
+
+
+@FastStreamBootstrapper.use_instrument()
+class FastStreamOpentelemetryInstrument(BaseOpentelemetryInstrument[FastStreamOpentelemetryConfig]):
+    def is_ready(self) -> bool:
+        return bool(self.instrument_config.opentelemetry_middleware_cls and super().is_ready())
+
+    def bootstrap_after(self, application: AsgiFastStream) -> AsgiFastStream:  # type: ignore[override]
+        if self.instrument_config.opentelemetry_middleware_cls and application.broker:
+            application.broker.add_middleware(
+                self.instrument_config.opentelemetry_middleware_cls(tracer_provider=self.tracer_provider)
+            )
+        return application
+
+    @classmethod
+    def get_config_type(cls) -> type[FastStreamOpentelemetryConfig]:
+        return FastStreamOpentelemetryConfig
+
+
+@FastStreamBootstrapper.use_instrument()
+class FastStreamLoggingInstrument(LoggingInstrument):
+    def bootstrap_before(self) -> dict[str, typing.Any]:
+        return {"logger": structlog.get_logger("microbootstrap-faststream")}
+
+
+@FastStreamBootstrapper.use_instrument()
+class FastStreamPrometheusInstrument(PrometheusInstrument[FastStreamPrometheusConfig]):
+    def is_ready(self) -> bool:
+        return bool(self.instrument_config.prometheus_middleware_cls and super().is_ready())
+
+    def bootstrap_before(self) -> dict[str, typing.Any]:
+        self.collector_registry = prometheus_client.CollectorRegistry()
+        return {
+            "asgi_routes": (
+                (
+                    self.instrument_config.prometheus_metrics_path,
+                    prometheus_client.make_asgi_app(self.collector_registry),
+                ),
+            )
+        }
+
+    def bootstrap_after(self, application: AsgiFastStream) -> AsgiFastStream:  # type: ignore[override]
+        if self.instrument_config.prometheus_middleware_cls and application.broker:
+            application.broker.add_middleware(
+                self.instrument_config.prometheus_middleware_cls(registry=self.collector_registry)
+            )
+        return application
+
+    @classmethod
+    def get_config_type(cls) -> type[FastStreamPrometheusConfig]:
+        return FastStreamPrometheusConfig
+
+
+@FastStreamBootstrapper.use_instrument()
+class FastStreamHealthChecksInstrument(HealthChecksInstrument):
+    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"]
+                else AsgiResponse(b"Service is unhealthy", 500, headers={"content-type": "application/json"})
+            )
+
+        return {"asgi_routes": ((self.instrument_config.health_checks_path, check_health),)}

microbootstrap/config/faststream.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+import dataclasses
+import typing
+
+
+if typing.TYPE_CHECKING:
+    import faststream.asyncapi.schema as asyncapi
+    from faststream.asgi.types import ASGIApp
+    from faststream.broker.core.usecase import BrokerUsecase
+    from faststream.types import AnyDict, AnyHttpUrl, Lifespan
+
+
+@dataclasses.dataclass
+class FastStreamConfig:
+    broker: BrokerUsecase[typing.Any, typing.Any] | None = None
+    asgi_routes: typing.Sequence[tuple[str, ASGIApp]] = ()
+    lifespan: Lifespan | None = None
+    terms_of_service: AnyHttpUrl | None = None
+    license: asyncapi.License | asyncapi.LicenseDict | AnyDict | None = None
+    contact: asyncapi.Contact | asyncapi.ContactDict | AnyDict | None = None
+    tags: typing.Sequence[asyncapi.Tag | asyncapi.TagDict | AnyDict] | None = None
+    external_docs: asyncapi.ExternalDocs | asyncapi.ExternalDocsDict | AnyDict | None = None
+    identifier: str | None = None
+    on_startup: typing.Sequence[typing.Callable[..., typing.Any]] = ()
+    after_startup: typing.Sequence[typing.Callable[..., typing.Any]] = ()
+    on_shutdown: typing.Sequence[typing.Callable[..., typing.Any]] = ()
+    after_shutdown: typing.Sequence[typing.Callable[..., typing.Any]] = ()