Initial working draft

Author: febus982

src/bootstrap/bootstrap.py

@@ -4,21 +4,22 @@
 from dependency_injector.containers import DynamicContainer
 from dependency_injector.providers import Object
 from faststream.broker.core.usecase import BrokerUsecase
+from faststream.redis import RedisBroker
 # 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 .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]
+    faststream_broker: RedisBroker
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
@@ -33,7 +34,7 @@ def application_init(app_config: AppConfig) -> InitReference:
     init_logger(app_config)
     init_storage()
     celery = init_celery(app_config)
-    # broker = init_broker(app_config)
+    broker = init_broker(app_config)
     # This is temporary, has to go directly in the Container
     # container.BookEventGatewayInterface.override(
     #     Object(FastStreamRedisGateway(broker=broker))
@@ -42,5 +43,5 @@ def application_init(app_config: AppConfig) -> InitReference:
     return InitReference(
         celery_app=celery,
         di_container=container,
-        # faststream_broker=broker,
+        faststream_broker=broker,
     )

src/bootstrap/config.py

@@ -51,7 +51,7 @@ class AppConfig(BaseSettings):
 
     APP_NAME: str = "bootstrap"
     CELERY: CeleryConfig = CeleryConfig()
-    # EVENTS: EventConfig
+    EVENTS: EventConfig
     DEBUG: bool = False
     ENVIRONMENT: TYPE_ENVIRONMENT = "local"
     SQLALCHEMY_CONFIG: Dict[str, SQLAlchemyConfig] = dict(

src/bootstrap/faststream.py

@@ -1,24 +1,24 @@
 # from typing import Dict, Union
-#
-# import structlog
-# # from domains.events import get_topic_registry
-# from faststream.redis import RedisBroker
+
+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
-#
-#
+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]:

src/event_consumer/__init__.py

@@ -3,7 +3,6 @@
 application layers (storage, logs) when running standalone workers
 without having to initialise the HTTP framework (or other ones)
 """
-
 import os
 from typing import Annotated, Union, Dict, Type, Optional, List
 
@@ -18,7 +17,7 @@
 from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import BatchSpanProcessor
 
-from domains.books.events import BookCreatedV1, BookUpdatedV1
+from domains.books.events import BookCreatedV1, BookUpdatedV1, BookCreatedV1Data
 
 _event_registry: Dict[str, Type] = {
     'books': Annotated[
@@ -36,24 +35,41 @@ def setup_telemetry(service_name: str, otlp_endpoint: str) -> TracerProvider:
     return tracer_provider
 
 
-def create_app(
-    test_config: Union[AppConfig, None] = None
-) -> FastStream:
-    setup_telemetry("faststream", otlp_endpoint=os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"])
-    broker = application_init(test_config or AppConfig()).faststream_broker
-    return FastStream(broker, logger=structlog.get_logger())
-
-
 def register_subscribers(broker, topics: Optional[List[str]] = None):
     if topics is None:
         topics_map: Dict[str, Type] = _event_registry
     else:
         topics_map: Dict[str, Type] = {k: v for k, v in _event_registry.items() if k in topics}
 
-    for topic, event_type in topics_map:
+
+    for topic, event_type in topics_map.items():
         @broker.subscriber(topic)  # type: ignore
         async def handler(msg: event_type, logger: Logger) -> None:
-            logger.info("Received message", extra={"msg": "some_extra_here"})
+            logger.info(f"Received message {type(msg)} {msg}")
+            # logger.info(f"Received message {type(msg)} {msg}", extra={"msg": "some_extra_here"})
             # l = logging.getLogger()
             # l.info("AAAAA", extra={"eee": "AAA"})
 
+
+def create_app(
+    test_config: Union[AppConfig, None] = None
+) -> FastStream:
+    setup_telemetry("faststream", otlp_endpoint=os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"])
+    broker = application_init(AppConfig()).faststream_broker
+    app = FastStream(broker, logger=structlog.get_logger())
+    register_subscribers(broker)
+
+    publisher = broker.publisher("books", schema=_event_registry["books"])
+
+    @app.after_startup
+    async def after_startup():
+        await broker.publish(BookCreatedV1.event_factory(
+            data=BookCreatedV1Data(
+                book_id=123,
+                title="AAA",
+                author_name="BBB",
+            )
+        ), "books")
+
+
+    return app