WIP commit

Author: febus982

docker-compose.yaml

@@ -36,12 +36,11 @@ services:
       context: .
       target: dev
     environment:
-      WATCHFILES_FORCE_POLLING: true
       OTEL_SERVICE_NAME: "bootstrap-fastapi-dev"
       OTEL_EXPORTER_OTLP_ENDPOINT: "http://otel-collector:4317"
       EVENTS__REDIS_BROKER_URL: "redis://redis:6379/2"
-      EVENTS__IS_PUBLISHER: "True"
-      EVENTS__IS_SUBSCRIBER: "True"
+      EVENTS__REGISTER_PUBLISHERS: "True"
+      EVENTS__REGISTER_SUBSCRIBERS: "True"
     working_dir: "/app/src"
     volumes:
       - '.:/app'
@@ -51,8 +50,9 @@ services:
     command:
       - faststream
       - run
-      - event_consumer:create_app
+#      - --help
       - --factory
+      - event_consumer:create_app
 
   event-docs:
     build:
@@ -64,8 +64,8 @@ services:
       OTEL_SERVICE_NAME: "bootstrap-fastapi-dev"
       OTEL_EXPORTER_OTLP_ENDPOINT: "http://otel-collector:4317"
       EVENTS__REDIS_BROKER_URL: "redis://redis:6379/2"
-      EVENTS__IS_PUBLISHER: "True"
-      EVENTS__IS_SUBSCRIBER: "True"
+      EVENTS__REGISTER_PUBLISHERS: "True"
+      EVENTS__REGISTER_SUBSCRIBERS: "True"
     ports:
       - '8000:8000'
     working_dir: "/app/src"
@@ -96,6 +96,7 @@ services:
       OTEL_SERVICE_NAME: "bootstrap-fastapi-http"
       OTEL_EXPORTER_OTLP_ENDPOINT: "http://otel-collector:4317"
       EVENTS__REDIS_BROKER_URL: "redis://redis:6379/2"
+      EVENTS__REGISTER_PUBLISHERS: "True"
     ports:
       - '8001:8000'
     volumes:

src/common/bootstrap.py

@@ -2,22 +2,22 @@
 
 from dependency_injector.containers import DynamicContainer
 from dependency_injector.providers import Object
-from faststream.redis import RedisBroker, RedisRouter
+from faststream.redis import RedisBroker
 
 # from gateways.event import FastStreamRedisGateway
 from pydantic import BaseModel, ConfigDict
 
 from .config import AppConfig
 from .di_container import Container
 from .dramatiq import init_dramatiq
-from .faststream import init_router
+from .faststream import init_broker
 from .logs import init_logger
 from .storage import init_storage
 
 
 class InitReference(BaseModel):
     di_container: DynamicContainer
-    faststream_broker: RedisRouter
+    faststream_broker: RedisBroker
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
@@ -32,13 +32,13 @@ def application_init(app_config: AppConfig) -> InitReference:
     init_logger(app_config)
     init_storage()
     init_dramatiq(app_config)
-    broker = init_router(app_config.EVENTS)
+    router = init_broker(app_config.EVENTS)
     # This is temporary, has to go directly in the Container
     # container.BookEventGatewayInterface.override(
     #     Object(FastStreamRedisGateway(broker=broker))
     # )
 
     return InitReference(
         di_container=container,
-        faststream_broker=broker,
+        faststream_broker=router,
     )

src/common/config.py

@@ -19,9 +19,9 @@ class AuthConfig(BaseModel):
 
 class EventConfig(BaseModel):
     REDIS_BROKER_URL: str = ""
-    TOPIC: Optional[str] = None
-    IS_PUBLISHER: bool = False
-    IS_SUBSCRIBER: bool = False
+    SUBSCRIBER_TOPIC: Optional[str] = None
+    REGISTER_PUBLISHERS: bool = False
+    REGISTER_SUBSCRIBERS: bool = False
 
 
 class AppConfig(BaseSettings):

src/common/faststream.py

@@ -2,37 +2,29 @@
 from typing import Optional
 
 import structlog
-from faststream import Logger
-
-# from domains.events import get_topic_registry
-from faststream.redis import RedisRouter, RedisBroker, fastapi
-
-# from faststream.redis.publisher.asyncapi import AsyncAPIPublisher
+from faststream.redis import RedisBroker, RedisRouter
 from opentelemetry.instrumentation.faststream import RedisOtelMiddleware
 
-from common.config import EventConfig
 from domains import event_registry
-from event_consumer import register_subscribers
+
+from .config import EventConfig
 
 logger = structlog.getLogger(__name__)
 
 
-def init_router(config: EventConfig) -> RedisRouter:
+def init_broker(config: EventConfig) -> RedisBroker:
     broker = RedisBroker(
         config.REDIS_BROKER_URL,
         middlewares=(RedisOtelMiddleware,),
         logger=structlog.getLogger("faststream.broker"),
     )
 
-    router = RedisRouter()
-    register_publishers(router, config.TOPIC)
-    if config.IS_SUBSCRIBER:
-        register_subscribers(router, config.TOPIC)
-
-    broker.include_router(router)
-    return router
-
+    if config.REGISTER_PUBLISHERS:
+        router = RedisRouter()
+        register_publishers(router, config.SUBSCRIBER_TOPIC)
+        broker.include_router(router)
 
+    return broker
 
 
 def register_publishers(router: RedisRouter, topic: Optional[str] = None):

src/event_consumer/__init__.py

@@ -5,22 +5,29 @@
 """
 
 import os
-from typing import Union, Optional
+from typing import Optional, Union
 
 import structlog
 from faststream import FastStream
-from faststream.redis import RedisRouter
+from faststream.redis import RedisBroker, RedisRouter
 from opentelemetry import trace
 from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import BatchSpanProcessor
 
-from alembic.env import logger
 from common import AppConfig, application_init
+from conftest import test_config
 from domains import event_registry
 from domains.books.events import BookCreatedV1, BookCreatedV1Data
 
+"""
+For the sake of this example app we reuse the domain registry,
+which is used for publishing. In a real world these registries
+are different and separate.
+"""
+subscriber_registry = event_registry
+
 
 def setup_telemetry(service_name: str, otlp_endpoint: str) -> TracerProvider:
     resource = Resource.create(attributes={"service.name": service_name})
@@ -33,11 +40,15 @@ def setup_telemetry(service_name: str, otlp_endpoint: str) -> TracerProvider:
 
 
 def create_app(test_config: Union[AppConfig, None] = None) -> FastStream:
+    config = test_config or AppConfig()
     setup_telemetry(
         "faststream", otlp_endpoint=os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"]
     )
-    router = application_init(AppConfig()).faststream_broker
-    broker = router.broker
+    broker = application_init(config).faststream_broker
+    register_subscribers(broker)
+    if config.EVENTS.REGISTER_SUBSCRIBERS:
+        register_subscribers(broker)
+
     app = FastStream(broker, logger=structlog.get_logger())
 
     @app.after_startup
@@ -56,18 +67,20 @@ async def after_startup():
     return app
 
 
-
-# TODO: Add Routing structure similar to the one in the fastapi and
-#       move this in the event_consumer_module
-def register_subscribers(router: RedisRouter, topic: Optional[str] = None):
-    if topic is not None and topic in event_registry.keys():
-        topics_map = {topic: event_registry[topic]}
+# TODO: Add Routing structure similar to the one in the fastapi implementation
+def register_subscribers(broker: RedisBroker, topic: Optional[str] = None):
+    if topic is not None and topic in subscriber_registry.keys():
+        topics_map = {topic: subscriber_registry[topic]}
     else:
-        topics_map = event_registry.copy()
+        topics_map = subscriber_registry.copy()
 
     logger = structlog.get_logger()
+    router = RedisRouter()
 
     for topic, event_type in topics_map.items():
+
         @router.subscriber(topic)
         async def handler(msg: event_type) -> None:  # type: ignore[valid-type]
             logger.info(f"Received message {type(msg)} {msg}")
+
+    broker.include_router(router)

src/http_app/__init__.py

@@ -1,4 +1,4 @@
-from typing import Union, Any
+from typing import Any, Union
 
 from fastapi import FastAPI, Request
 from faststream.broker.core.usecase import BrokerUsecase
@@ -36,7 +36,7 @@ def create_app(
     init_exception_handlers(app)
 
     init_routes(app)
-    add_faststream_router(app, ref.faststream_broker)
+    # add_faststream_router(app, ref.faststream_broker)
 
     """
     OpenTelemetry prometheus exporter does not work together with automatic