Init dramatiq on application start

Signed-off-by: Federico Busetti <[email protected]>

Author: febus982

src/common/bootstrap.py

@@ -8,6 +8,7 @@
 from .celery import init_celery
 from .config import AppConfig
 from .di_container import Container
+from .dramatiq import init_dramatiq
 from .logs import init_logger
 from .storage import init_storage
 
@@ -28,6 +29,7 @@ def application_init(app_config: AppConfig) -> InitReference:
     )
     init_logger(app_config)
     init_storage()
+    init_dramatiq(app_config)
     celery = init_celery(app_config)
 
     return InitReference(

src/common/dramatiq.py

@@ -0,0 +1,17 @@
+from dramatiq import set_broker
+from dramatiq.brokers.stub import StubBroker
+from dramatiq.middleware import AsyncIO
+from opentelemetry_instrumentor_dramatiq import DramatiqInstrumentor
+
+from .config import AppConfig
+
+
+def init_dramatiq(config: AppConfig):
+    DramatiqInstrumentor().instrument()
+    if config.ENVIRONMENT == "test":
+        broker = StubBroker()
+    else:
+        broker = StubBroker()
+    # broker.emit_after("process_boot")
+    broker.add_middleware(AsyncIO())
+    set_broker(broker)