WIP

Author: febus982

docker-compose.yaml

@@ -67,6 +67,8 @@ services:
       CELERY__broker_url: "redis://redis:6379/0"
       CELERY__result_backend: "redis://redis:6379/1"
       EVENTS__REDIS_BROKER_URL: "redis://redis:6379/2"
+      EVENTS__IS_PUBLISHER: "True"
+      EVENTS__IS_SUBSCRIBER: "False"
     ports:
       - '8000:8000'
     working_dir: "/app/src"
@@ -99,6 +101,8 @@ services:
       CELERY__broker_url: "redis://redis:6379/0"
       CELERY__result_backend: "redis://redis:6379/1"
       EVENTS__REDIS_BROKER_URL: "redis://redis:6379/2"
+      EVENTS__IS_PUBLISHER: "True"
+      EVENTS__IS_SUBSCRIBER: "True"
     working_dir: "/app/src"
     volumes:
       - '.:/app'
@@ -108,7 +112,6 @@ services:
     command:
       - faststream
       - run
-#      - event_consumer:app
       - event_consumer:create_app
       - --factory
 
@@ -124,6 +127,8 @@ services:
       CELERY__broker_url: "redis://redis:6379/0"
       CELERY__result_backend: "redis://redis:6379/1"
       EVENTS__REDIS_BROKER_URL: "redis://redis:6379/2"
+      EVENTS__IS_PUBLISHER: "True"
+      EVENTS__IS_SUBSCRIBER: "True"
     ports:
       - '8000:8000'
     working_dir: "/app/src"

src/bootstrap/bootstrap.py

@@ -3,23 +3,23 @@
 from celery import Celery
 from dependency_injector.containers import DynamicContainer
 from dependency_injector.providers import Object
-from faststream.redis import RedisBroker
+from faststream.redis import RedisBroker, RedisRouter
 
 # 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_router
 from .logs import init_logger
 from .storage import init_storage
 
 
 class InitReference(BaseModel):
     celery_app: Celery
     di_container: DynamicContainer
-    faststream_broker: RedisBroker
+    faststream_broker: RedisRouter
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
@@ -34,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_router(app_config.EVENTS)
     # This is temporary, has to go directly in the Container
     # container.BookEventGatewayInterface.override(
     #     Object(FastStreamRedisGateway(broker=broker))

src/bootstrap/config.py

@@ -49,6 +49,8 @@ class CeleryConfig(BaseModel):
 class EventConfig(BaseModel):
     REDIS_BROKER_URL: str = ""
     TOPIC: Optional[str] = None
+    IS_PUBLISHER: bool = False
+    IS_SUBSCRIBER: bool = False
 
 
 class AppConfig(BaseSettings):

src/bootstrap/faststream.py

@@ -5,55 +5,40 @@
 from faststream import Logger
 
 # from domains.events import get_topic_registry
-from faststream.redis import RedisBroker
+from faststream.redis import RedisRouter, RedisBroker, fastapi
 
 # from faststream.redis.publisher.asyncapi import AsyncAPIPublisher
 from opentelemetry.instrumentation.faststream import RedisOtelMiddleware
 
+from bootstrap.config import EventConfig
 from domains import event_registry
 
-from .config import AppConfig
+logger = structlog.getLogger(__name__)
 
 
-def init_broker(config: AppConfig) -> RedisBroker:
+def init_router(config: EventConfig) -> RedisRouter:
     broker = RedisBroker(
-        config.EVENTS.REDIS_BROKER_URL,
+        config.REDIS_BROKER_URL,
         middlewares=(RedisOtelMiddleware,),
         logger=structlog.getLogger("faststream.broker"),
     )
-    register_publishers(broker, config.EVENTS.TOPIC)
-    register_subscribers(broker, config.EVENTS.TOPIC)
 
-    return broker
+    router = RedisRouter()
+    register_publishers(router, config.TOPIC)
+    if config.IS_SUBSCRIBER:
+        register_subscribers(router, config.TOPIC)
 
+    broker.include_router(router)
+    return router
 
-def register_subscribers(broker, topic: Optional[str] = None):
-    if topic is not None and topic in event_registry.keys():
-        topics_map = {topic: event_registry[topic]}
-    else:
-        topics_map = event_registry.copy()
 
-    for topic, event_type in topics_map.items():
-
-        @broker.subscriber(topic)
-        async def handler(msg: event_type, logger: Logger) -> None:  # type: ignore[valid-type]
-            logger.info(f"Received message {type(msg)} {msg}")
 
 
-def register_publishers(broker, topic: Optional[str] = None):
+def register_publishers(router: RedisRouter, topic: Optional[str] = None):
     if topic is not None and topic in event_registry.keys():
         topics_map = {topic: event_registry[topic]}
     else:
         topics_map = event_registry.copy()
 
     for topic, event_type in topics_map.items():
-        broker.publisher(topic, schema=event_registry[topic])
-
-
-# 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()
-#     }
+        router.publisher(topic, schema=event_registry[topic])

src/event_consumer/__init__.py

@@ -5,17 +5,20 @@
 """
 
 import os
-from typing import Union
+from typing import Union, Optional
 
 import structlog
 from faststream import FastStream
+from faststream.redis import 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 bootstrap import AppConfig, application_init
+from domains import event_registry
 from domains.books.events import BookCreatedV1, BookCreatedV1Data
 
 
@@ -33,7 +36,8 @@ 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
+    router = application_init(AppConfig()).faststream_broker
+    broker = router.broker
     app = FastStream(broker, logger=structlog.get_logger())
 
     @app.after_startup
@@ -50,3 +54,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]}
+    else:
+        topics_map = event_registry.copy()
+
+    logger = structlog.get_logger()
+
+    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}")

src/http_app/__init__.py

@@ -1,9 +1,8 @@
-from contextlib import asynccontextmanager
-from typing import Any, Union
+from typing import Union, Any
 
 from fastapi import FastAPI, Request
 from faststream.broker.core.usecase import BrokerUsecase
-from faststream.types import Lifespan
+from faststream.redis import RedisRouter, fastapi
 from opentelemetry.instrumentation.asgi import OpenTelemetryMiddleware
 from starlette.responses import JSONResponse
 from starlette_prometheus import PrometheusMiddleware, metrics
@@ -19,20 +18,14 @@ def create_app(
     app_config = test_config or AppConfig()
     ref = application_init(app_config)
 
-    """
-    We don't want to couple together FastAPI and FastStream
-    so we use a lifespan handler instead of the plugin to
-    remain decoupled but still able use any generic broker
-    for message publishing in FastAPI.
-    """
     app = FastAPI(
         debug=app_config.DEBUG,
         title=app_config.APP_NAME,
-        lifespan=faststream_lifespan(ref.faststream_broker),
     )
     init_exception_handlers(app)
 
     init_routes(app)
+    add_faststream_router(app, ref.faststream_broker)
 
     """
     OpenTelemetry prometheus exporter does not work together with automatic
@@ -67,13 +60,7 @@ async def add_exception_middleware(request: Request, call_next):
             return JSONResponse({"error": "Internal server error"}, status_code=500)
 
 
-def faststream_lifespan(broker: BrokerUsecase[Any, Any]) -> Lifespan:
-    @asynccontextmanager
-    async def handler(app: FastAPI):
-        await broker.start()
-        try:
-            yield
-        finally:
-            await broker.close()
-
-    return handler
+def add_faststream_router(app: FastAPI, router: RedisRouter) -> None:
+    f_router = fastapi.RedisRouter()
+    f_router.include_router(router)
+    app.include_router(f_router)

src/http_app/routes/ping.py

@@ -1,5 +1,6 @@
 from fastapi import APIRouter
 from pydantic import BaseModel, ConfigDict
+from starlette.responses import JSONResponse
 
 router = APIRouter()
 
@@ -18,4 +19,5 @@ class PingResponse(BaseModel):
 
 @router.get("/ping")
 async def ping() -> PingResponse:
+    JSONResponse({"ping": "pong!"})
     return PingResponse(ping="pong!")