WIP

Signed-off-by: Federico Busetti <729029+febus982@users.noreply.github.com>

Author: febus982

docker-compose.yaml

@@ -7,6 +7,8 @@ services:
 
   redis:
     image: redis
+    ports:
+      - "6379:6379"
 
   celery-worker:
     build:

poetry.lock

@@ -25,15 +25,17 @@ asgiref = "^3.7.2"
 celery =  { version = "^5.3.1", extras = ["redis"] }
 cloudevents-pydantic = "^0.0.2"
 dependency-injector = { version = "^4.41.0", extras = ["pydantic"] }
+faststream = { version = "^0.5.25", extras = ["redis"] }
 httpx = ">=0.23.0"
 opentelemetry-distro = { version = "*", extras = ["otlp"] }
 opentelemetry-instrumentation = "*"
 opentelemetry-instrumentation-celery = "*"
+opentelemetry-instrumentation-faststream = "*"
 opentelemetry-instrumentation-httpx = "*"
 opentelemetry-instrumentation-sqlalchemy = "*"
 pydantic = "^2.2.1"
 pydantic-settings = "^2.0.3"
-python = ">=3.9,<3.13"
+python = ">=3.11,<3.13"
 rich = "^13.2.0"
 SQLAlchemy = { version = "^2.0.0", extras = ["asyncio", "mypy"] }
 sqlalchemy-bind-manager = "*"

pyproject.toml

@@ -1,20 +1,24 @@
-from typing import cast
+from typing import cast, Any
 
 from celery import Celery
 from dependency_injector.containers import DynamicContainer
 from dependency_injector.providers import Object
+from faststream.broker.core.usecase import BrokerUsecase
+# from gateways.event import FastStreamRedisGateway
 from pydantic import BaseModel, ConfigDict
 
 from .celery import init_celery
 from .config import AppConfig
 from .di_container import Container
+# from .faststream import init_broker
 from .logs import init_logger
 from .storage import init_storage
 
 
 class InitReference(BaseModel):
     celery_app: Celery
     di_container: DynamicContainer
+    # faststream_broker: BrokerUsecase[Any, Any]
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
@@ -29,8 +33,14 @@ def application_init(app_config: AppConfig) -> InitReference:
     init_logger(app_config)
     init_storage()
     celery = init_celery(app_config)
+    # broker = init_broker(app_config)
+    # This is temporary, has to go directly in the Container
+    # container.BookEventGatewayInterface.override(
+    #     Object(FastStreamRedisGateway(broker=broker))
+    # )
 
     return InitReference(
         celery_app=celery,
         di_container=container,
+        # faststream_broker=broker,
     )

src/bootstrap/bootstrap.py

@@ -42,11 +42,16 @@ class CeleryConfig(BaseModel):
     # }
 
 
+class EventConfig(BaseModel):
+    REDIS_BROKER_URL: str
+
+
 class AppConfig(BaseSettings):
     model_config = SettingsConfigDict(env_nested_delimiter="__")
 
     APP_NAME: str = "bootstrap"
     CELERY: CeleryConfig = CeleryConfig()
+    # EVENTS: EventConfig
     DEBUG: bool = False
     ENVIRONMENT: TYPE_ENVIRONMENT = "local"
     SQLALCHEMY_CONFIG: Dict[str, SQLAlchemyConfig] = dict(

src/bootstrap/config.py

@@ -0,0 +1,28 @@
+# from typing import Dict, Union
+#
+# import structlog
+# # from domains.events import get_topic_registry
+# from faststream.redis import RedisBroker
+# from faststream.redis.publisher.asyncapi import AsyncAPIPublisher
+# from opentelemetry.instrumentation.faststream import RedisOtelMiddleware
+#
+# from .config import AppConfig
+#
+#
+# def init_broker(config: AppConfig) -> RedisBroker:
+#     broker = RedisBroker(
+#         config.EVENTS.REDIS_BROKER_URL,
+#         middlewares=(RedisOtelMiddleware,),
+#         logger=structlog.getLogger("faststream.broker"),
+#     )
+#
+#     return broker
+#
+#
+# def init_publishers(
+#     broker: RedisBroker,
+# ) -> Dict[str, AsyncAPIPublisher]:
+#     return {
+#         topic: broker.publisher(topic, schema=Union[event_types])
+#         for topic, event_types in get_topic_registry().items()
+#     }