Eventy consumer refactor

Author: febus982

docker-compose.yaml

@@ -20,6 +20,7 @@ services:
       OTEL_EXPORTER_OTLP_ENDPOINT: "http://otel-collector:4317"
       CELERY__broker_url: "redis://redis:6379/0"
       CELERY__result_backend: "redis://redis:6379/1"
+      EVENTS__REDIS_BROKER_URL: "redis://redis:6379/2"
     working_dir: "/app/src"
     volumes:
       - '.:/app'
@@ -38,6 +39,7 @@ services:
       OTEL_EXPORTER_OTLP_ENDPOINT: "http://otel-collector:4317"
       CELERY__broker_url: "redis://redis:6379/0"
       CELERY__result_backend: "redis://redis:6379/1"
+      EVENTS__REDIS_BROKER_URL: "redis://redis:6379/2"
     working_dir: "/app/src"
     volumes:
       - '.:/app'
@@ -64,6 +66,7 @@ services:
       OTEL_EXPORTER_OTLP_ENDPOINT: "http://otel-collector:4317"
       CELERY__broker_url: "redis://redis:6379/0"
       CELERY__result_backend: "redis://redis:6379/1"
+      EVENTS__REDIS_BROKER_URL: "redis://redis:6379/2"
     ports:
       - '8000:8000'
     working_dir: "/app/src"
@@ -82,7 +85,61 @@ services:
       - "8000"
       - --factory
       # Remember to disable the reloader in order to allow otel instrumentation
-      - --reload
+#      - --reload
+
+  event-consumer:
+    build:
+      dockerfile: Dockerfile
+      context: .
+      target: dev
+    environment:
+      WATCHFILES_FORCE_POLLING: true
+      OTEL_SERVICE_NAME: "bootstrap-fastapi-dev"
+      OTEL_EXPORTER_OTLP_ENDPOINT: "http://otel-collector:4317"
+      CELERY__broker_url: "redis://redis:6379/0"
+      CELERY__result_backend: "redis://redis:6379/1"
+      EVENTS__REDIS_BROKER_URL: "redis://redis:6379/2"
+    working_dir: "/app/src"
+    volumes:
+      - '.:/app'
+    depends_on:
+      - redis
+      - otel-collector
+    command:
+      - faststream
+      - run
+#      - event_consumer:app
+      - event_consumer:create_app
+      - --factory
+
+  event-docs:
+    build:
+      dockerfile: Dockerfile
+      context: .
+      target: dev
+    environment:
+      WATCHFILES_FORCE_POLLING: true
+      OTEL_SERVICE_NAME: "bootstrap-fastapi-dev"
+      OTEL_EXPORTER_OTLP_ENDPOINT: "http://otel-collector:4317"
+      CELERY__broker_url: "redis://redis:6379/0"
+      CELERY__result_backend: "redis://redis:6379/1"
+      EVENTS__REDIS_BROKER_URL: "redis://redis:6379/2"
+    ports:
+      - '8000:8000'
+    working_dir: "/app/src"
+    volumes:
+      - '.:/app'
+    depends_on:
+      - redis
+      - otel-collector
+    command:
+      - faststream
+      - docs
+      - serve
+      - event_consumer:create_app
+      - --factory
+      - --host
+      - 0.0.0.0
 
   http:
     build:
@@ -97,6 +154,7 @@ services:
       OTEL_EXPORTER_OTLP_ENDPOINT: "http://otel-collector:4317"
       CELERY__broker_url: "redis://redis:6379/0"
       CELERY__result_backend: "redis://redis:6379/1"
+      EVENTS__REDIS_BROKER_URL: "redis://redis:6379/2"
     ports:
       - '8001:8000'
     volumes:

poetry.lock

@@ -23,7 +23,8 @@ aiosqlite = ">=0.18.0"
 alembic = "^1.11.1"
 asgiref = "^3.7.2"
 celery =  { version = "^5.3.1", extras = ["redis"] }
-cloudevents-pydantic = "^0.0.2"
+cloudevents-pydantic = { git = "https://github.com/febus982/cloudevents-pydantic", branch = "main" }
+#cloudevents-pydantic = "^0.0.2"
 dependency-injector = { version = "^4.41.0", extras = ["pydantic"] }
 faststream = { version = "^0.5.25", extras = ["redis"] }
 httpx = ">=0.23.0"
@@ -99,9 +100,7 @@ exclude_also = [
 [tool.mypy]
 files = ["src", "tests"]
 exclude = ["alembic"]
-# Pydantic plugin causes some issues: https://github.com/pydantic/pydantic-settings/issues/403
-#plugins = "pydantic.mypy,strawberry.ext.mypy_plugin"
-plugins = "strawberry.ext.mypy_plugin"
+plugins = "pydantic.mypy,strawberry.ext.mypy_plugin"
 python_version = "3.9"
 
 # We can remove celery by installing `celery-types` but needs

pyproject.toml

@@ -1,10 +1,10 @@
-from typing import cast, Any
+from typing import cast
 
 from celery import Celery
 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
 

src/bootstrap/bootstrap.py

@@ -1,3 +1,7 @@
+# mypy: disable-error-code="call-arg,syntax"
+# `call-arg` is because of nested models (they have to be supplied via ENV)
+# `syntax` is because of Pydantic plugin
+# https://github.com/pydantic/pydantic-settings/issues/403
 from pathlib import Path
 from typing import Dict, Literal, Optional
 
@@ -43,15 +47,20 @@ class CeleryConfig(BaseModel):
 
 
 class EventConfig(BaseModel):
-    REDIS_BROKER_URL: str
+    REDIS_BROKER_URL: str = ""
+    TOPIC: Optional[str] = None
 
 
 class AppConfig(BaseSettings):
-    model_config = SettingsConfigDict(env_nested_delimiter="__")
+    model_config = SettingsConfigDict(
+        case_sensitive=True,
+        env_nested_delimiter="__",
+        nested_model_default_partial_update=True,
+    )
 
     APP_NAME: str = "bootstrap"
     CELERY: CeleryConfig = CeleryConfig()
-    EVENTS: EventConfig
+    EVENTS: EventConfig = EventConfig()
     DEBUG: bool = False
     ENVIRONMENT: TYPE_ENVIRONMENT = "local"
     SQLALCHEMY_CONFIG: Dict[str, SQLAlchemyConfig] = dict(

src/bootstrap/config.py

@@ -1,11 +1,17 @@
 # from typing import Dict, Union
+from typing import Optional
 
 import structlog
+from faststream import Logger
+
 # 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 domains import event_registry
+
 from .config import AppConfig
 
 
@@ -15,10 +21,35 @@ def init_broker(config: AppConfig) -> RedisBroker:
         middlewares=(RedisOtelMiddleware,),
         logger=structlog.getLogger("faststream.broker"),
     )
+    register_publishers(broker, config.EVENTS.TOPIC)
+    register_subscribers(broker, config.EVENTS.TOPIC)
 
     return broker
 
 
+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):
+    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]:

src/bootstrap/faststream.py

@@ -0,0 +1 @@
+from ._event_registry import event_registry

src/domains/__init__.py

@@ -0,0 +1,10 @@
+from typing import Annotated, Dict, Union
+
+from pydantic import Field
+from typing_extensions import TypeAlias
+
+from .books.events import BookCreatedV1, BookUpdatedV1
+
+event_registry: Dict[str, TypeAlias] = {
+    "books": Annotated[Union[BookCreatedV1, BookUpdatedV1], Field(discriminator="type")]
+}

src/domains/_event_registry.py

@@ -3,27 +3,21 @@
 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
+from typing import Union
 
 import structlog
-from pydantic import Field
-
-from bootstrap import AppConfig, application_init
-from faststream import FastStream, Logger
+from faststream import FastStream
 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 domains.books.events import BookCreatedV1, BookUpdatedV1, BookCreatedV1Data
+from bootstrap import AppConfig, application_init
+from domains.books.events import BookCreatedV1, BookCreatedV1Data
 
-_event_registry: Dict[str, Type] = {
-    'books': Annotated[
-        Union[BookCreatedV1, BookUpdatedV1], Field(discriminator="type")
-    ]
-}
 
 def setup_telemetry(service_name: str, otlp_endpoint: str) -> TracerProvider:
     resource = Resource.create(attributes={"service.name": service_name})
@@ -35,41 +29,24 @@ def setup_telemetry(service_name: str, otlp_endpoint: str) -> TracerProvider:
     return tracer_provider
 
 
-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.items():
-        @broker.subscriber(topic)  # type: ignore
-        async def handler(msg: event_type, logger: Logger) -> None:
-            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"])
+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")
-
+        await broker.publish(
+            BookCreatedV1.event_factory(
+                data=BookCreatedV1Data(
+                    book_id=123,
+                    title="AAA",
+                    author_name="BBB",
+                )
+            ),
+            "books",
+        )
 
     return app