Remove faststream references

Author: febus982

pyproject.toml

@@ -21,15 +21,13 @@ dependencies = [
     "opentelemetry-instrumentation-httpx",
     "opentelemetry-instrumentation-sqlalchemy",
     "opentelemetry-instrumentor-dramatiq",
-    "opentelemetry-instrumentation-faststream",
     "orjson<4.0.0,>=3.10.12",
     "pydantic<3.0.0,>=2.2.1",
     "pydantic-settings<3.0.0,>=2.0.3",
     "rich<14.0.0,>=13.2.0",
     "SQLAlchemy[asyncio,mypy]<3.0.0,>=2.0.0",
     "sqlalchemy-bind-manager",
     "structlog<25.1.1,>=25.1.0",
-    "faststream>=0.5.34",
 ]
 
 [dependency-groups]

src/common/bootstrap.py

@@ -2,22 +2,20 @@
 
 from dependency_injector.containers import DynamicContainer
 from dependency_injector.providers import Object
-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_broker
+from .event_publisher import init_broker
 from .logs import init_logger
 from .storage import init_storage
 
 
 class InitReference(BaseModel):
     di_container: DynamicContainer
-    faststream_broker: RedisBroker
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
 

src/common/event_publisher.py

@@ -0,0 +1,24 @@
+# from typing import Dict, Union
+from typing import Optional
+
+import structlog
+
+from domains import event_registry
+
+from .config import EventConfig
+
+logger = structlog.getLogger(__name__)
+
+
+def init_broker(config: EventConfig):
+
+    if config.REGISTER_PUBLISHERS:
+        register_publishers(config.SUBSCRIBER_TOPIC)
+
+
+
+def register_publishers(topic: Optional[str] = None):
+    if topic is not None and topic in event_registry.keys():
+        logger.info(f"Registering publishers for topic {topic}")
+    else:
+        logger.info(f"Registering publishers for all topics")

src/common/faststream.py

@@ -1,4 +1,3 @@
-from faststream import context
 from opentelemetry import trace
 from structlog.typing import EventDict
 
@@ -13,29 +12,6 @@ def extract_from_record(_, __, event_dict: EventDict) -> EventDict:
     return event_dict
 
 
-def faststream_context(_, __, event_dict: EventDict) -> EventDict:
-    """
-    Extract FastStream context information and adds them to the event dict.
-    """
-    c = context.get_local("log_context") or {}
-    event_context = event_dict.get(
-        "event_context",
-        c.copy(),
-    )
-
-    # Handle undesired extra override from FastStream
-    extra = event_dict.get("extra", {}).copy()
-    if {"channel", "message_id"} == set(extra.keys()):
-        event_context["channel"] = extra["channel"]
-        event_context["message_id"] = extra["message_id"]
-        del event_dict["extra"]
-
-    if event_context:
-        event_dict["event_context"] = event_context
-
-    return event_dict
-
-
 def drop_color_message_key(_, __, event_dict: EventDict) -> EventDict:
     """
     Uvicorn logs the message a second time in the extra `color_message`, but we don't

src/common/logs/processors.py

@@ -8,13 +8,6 @@
 from typing import Optional, Union
 
 import structlog
-from faststream import FastStream
-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 common import AppConfig, application_init
 from conftest import test_config
@@ -29,58 +22,58 @@
 subscriber_registry = event_registry
 
 
-def setup_telemetry(service_name: str, otlp_endpoint: str) -> TracerProvider:
-    resource = Resource.create(attributes={"service.name": service_name})
-    tracer_provider = TracerProvider(resource=resource)
-    exporter = OTLPSpanExporter(endpoint=otlp_endpoint)
-    processor = BatchSpanProcessor(exporter)
-    tracer_provider.add_span_processor(processor)
-    trace.set_tracer_provider(tracer_provider)
-    return tracer_provider
+# def setup_telemetry(service_name: str, otlp_endpoint: str) -> TracerProvider:
+#     resource = Resource.create(attributes={"service.name": service_name})
+#     tracer_provider = TracerProvider(resource=resource)
+#     exporter = OTLPSpanExporter(endpoint=otlp_endpoint)
+#     processor = BatchSpanProcessor(exporter)
+#     tracer_provider.add_span_processor(processor)
+#     trace.set_tracer_provider(tracer_provider)
+#     return tracer_provider
 
 
-def create_app(test_config: Union[AppConfig, None] = None) -> FastStream:
+def create_app(test_config: Union[AppConfig, None] = None):
     config = test_config or AppConfig()
-    setup_telemetry(
-        "faststream", otlp_endpoint=os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"]
-    )
-    broker = application_init(config).faststream_broker
-    register_subscribers(broker)
+    # setup_telemetry(
+    #     "faststream", otlp_endpoint=os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"]
+    # )
+    ref = application_init(config)
+    # register_subscribers(broker)
     if config.EVENTS.REGISTER_SUBSCRIBERS:
-        register_subscribers(broker)
-
-    app = FastStream(broker, logger=structlog.get_logger())
-
-    @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
+        register_subscribers(config.EVENTS.SUBSCRIBER_TOPIC)
+
+    # app = FastStream(broker, logger=structlog.get_logger())
+    #
+    # @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
 
 
 # TODO: Add Routing structure similar to the one in the fastapi implementation
-def register_subscribers(broker: RedisBroker, topic: Optional[str] = None):
+def register_subscribers(topic: Optional[str] = None):
     if topic is not None and topic in subscriber_registry.keys():
         topics_map = {topic: subscriber_registry[topic]}
     else:
         topics_map = subscriber_registry.copy()
 
     logger = structlog.get_logger()
-    router = RedisRouter()
 
     for topic, event_type in topics_map.items():
+        logger.info(f"Registering {event_type} on topic {topic}")
 
-        @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)
+    #     @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/event_consumer/__init__.py

@@ -11,32 +11,3 @@ async def emit(
             "Event emitted",
             cloudevent=event.model_dump(),
         )
-
-
-# class FastStreamRedisGateway:
-#     _broker: RedisBroker
-#     _publishers: Dict[type[BaseEvent], AsyncAPIPublisher]
-#
-#     def __init__(
-#         self,
-#         broker: RedisBroker,
-#         topic_filter: Optional[Collection[str]] = None,
-#     ):
-#         self._broker = broker
-#         publishers = {
-#             topic: broker.publisher(topic, schema=Union[event_types])
-#             for topic, event_types in get_topic_registry(topic_filter).items()
-#         }
-#         self._publishers = {
-#             event_type: publishers[topic]
-#             for topic, event_types in get_topic_registry(topic_filter).items()
-#             for event_type in event_types
-#         }
-#
-#     async def emit(
-#         self, event: BaseEvent
-#     ) -> None:  # pragma: no cover # No need to test this
-#         try:
-#             await self._publishers[type(event)].publish(event)
-#         except KeyError:
-#             raise RuntimeError(f"Unknown event type: {type(event)}")

src/gateways/event.py

@@ -1,8 +1,6 @@
 from typing import Any, Union
 
 from fastapi import FastAPI, Request
-from faststream.broker.core.usecase import BrokerUsecase
-from faststream.redis import RedisRouter, fastapi
 from opentelemetry.instrumentation.asgi import OpenTelemetryMiddleware
 from starlette.responses import JSONResponse
 from starlette_prometheus import PrometheusMiddleware, metrics
@@ -69,9 +67,3 @@ async def add_exception_middleware(request: Request, call_next):
             logger = get_logger(__name__)
             await logger.aexception(e)
             return JSONResponse({"error": "Internal server error"}, status_code=500)
-
-
-def add_faststream_router(app: FastAPI, router: RedisRouter) -> None:
-    f_router = fastapi.RedisRouter()
-    f_router.include_router(router)
-    app.include_router(f_router)