Feat/faststream structlog logger

docs/integrations/faststream.md

@@ -26,6 +26,7 @@ Read more about available extras [here](../../../introduction/installation):
 
 ```python
 from lite_bootstrap import FastStreamConfig, FastStreamBootstrapper
+from faststream.asgi import AsgiFastStream
 from faststream.redis.opentelemetry import RedisTelemetryMiddleware
 from faststream.redis.prometheus import RedisPrometheusMiddleware
 from faststream.redis import RedisBroker
@@ -44,7 +45,7 @@ bootstrapper_config = FastStreamConfig(
     sentry_dsn="https://testdsn@localhost/1",
     health_checks_path="/custom-health/",
     logging_buffer_capacity=0,
-    broker=broker,
+    application=AsgiFastStream(broker),
 )
 bootstrapper = FastStreamBootstrapper(bootstrapper_config)
 application = bootstrapper.bootstrap()

docs/introduction/configuration.md

@@ -115,6 +115,28 @@ Additional parameters:
 - `logging_extra_processors`
 - `logging_unset_handlers`.
 
+### Structlog FastStream
+
+When using FastStream, the structlog logger is automatically injected into the broker so that all broker
+service messages (e.g. "Received", "Processed") are routed through structlog.
+
+The broker log level is controlled independently from the application log level:
+
+- `faststream_log_level` - log level for FastStream broker service messages (default: `logging.WARNING`).
+
+This allows you to suppress broker noise while keeping your application logs at a lower level:
+
+```python
+import logging
+from lite_bootstrap import FastStreamConfig
+
+config = FastStreamConfig(
+    service_debug=False,
+    logging_log_level=logging.INFO,       # your application logs
+    faststream_log_level=logging.WARNING, # broker "Received"/"Processed" messages (default)
+)
+```
+
 ## CORS
 
 To bootstrap CORS headers, you must provide `cors_allowed_origins` or `cors_allowed_origin_regex`.

lite_bootstrap/bootstrappers/faststream_bootstrapper.py

@@ -1,5 +1,6 @@
 import dataclasses
 import json
+import logging
 import typing
 
 from lite_bootstrap import import_checker
@@ -13,9 +14,13 @@
 
 
 if import_checker.is_faststream_installed:
+    from faststream._internal.logger.params_storage import ManualLoggerStorage
     from faststream.asgi import AsgiFastStream, AsgiResponse
     from faststream.asgi import get as handle_get
 
+if import_checker.is_structlog_installed:
+    import structlog
+
 if import_checker.is_prometheus_client_installed:
     import prometheus_client
 
@@ -62,6 +67,7 @@ class FastStreamConfig(
     application: "AsgiFastStream" = dataclasses.field(default_factory=_make_asgi_faststream)
     opentelemetry_middleware_cls: type[FastStreamTelemetryMiddlewareProtocol] | None = None
     prometheus_middleware_cls: type[FastStreamPrometheusMiddlewareProtocol] | None = None
+    faststream_log_level: int = logging.WARNING
 
 
 @dataclasses.dataclass(kw_only=True, slots=True, frozen=True)
@@ -99,6 +105,13 @@ async def _define_health_status(self) -> bool:
 class FastStreamLoggingInstrument(LoggingInstrument):
     bootstrap_config: FastStreamConfig
 
+    def bootstrap(self) -> None:
+        super().bootstrap()
+        broker = self.bootstrap_config.application.broker
+        if broker is not None and import_checker.is_structlog_installed and import_checker.is_faststream_installed:
+            broker.config.logger.params_storage = ManualLoggerStorage(structlog.get_logger("faststream"))
+            broker.config.logger.set_level(self.bootstrap_config.faststream_log_level)
+
 
 @dataclasses.dataclass(kw_only=True, frozen=True)
 class FastStreamOpenTelemetryInstrument(OpenTelemetryInstrument):

tests/test_faststream_bootstrap.py

@@ -1,9 +1,11 @@
+import logging
 import typing
 
 import faststream.asgi
 import pytest
 import structlog
 from faststream._internal.broker import BrokerUsecase
+from faststream._internal.logger.params_storage import ManualLoggerStorage
 from faststream.redis import RedisBroker, TestRedisBroker
 from faststream.redis.opentelemetry import RedisTelemetryMiddleware
 from faststream.redis.prometheus import RedisPrometheusMiddleware
@@ -85,6 +87,23 @@ async def test_faststream_bootstrap_health_check_wo_broker() -> None:
     bootstrapper.teardown()
 
 
+def test_faststream_logging_instrument_injects_structlog_logger(broker: RedisBroker) -> None:
+    bootstrap_config = FastStreamConfig(
+        service_debug=False,
+        logging_buffer_capacity=0,
+        logging_log_level=logging.INFO,
+        faststream_log_level=logging.WARNING,
+        application=faststream.asgi.AsgiFastStream(broker),
+    )
+    bootstrapper = FastStreamBootstrapper(bootstrap_config=bootstrap_config)
+    bootstrapper.bootstrap()
+    try:
+        assert isinstance(broker.config.logger.params_storage, ManualLoggerStorage)
+        assert broker.config.logger.log_level == logging.WARNING
+    finally:
+        bootstrapper.teardown()
+
+
 def test_faststream_config_default_application() -> None:
     config = FastStreamConfig()
     assert isinstance(config.application, faststream.asgi.AsgiFastStream)