Rabbit sync new

propan/__init__.py

@@ -12,10 +12,10 @@
 else:
     RabbitBroker = RabbitRouter = about.INSTALL_RABBIT  # type: ignore
 
-if is_installed("pika"):
+try:
     from propan.brokers.rabbit.rabbit_broker_sync import RabbitSyncBroker
     from propan.brokers.rabbit.routing import RabbitRouter
-else:
+except ImportError:
     RabbitSyncBroker = RabbitRouter = about.INSTALL_RABBIT_SYNC  # type: ignore
 
 if is_installed("nats"):

propan/brokers/_model/__init__.py

@@ -1,10 +1,11 @@
-from propan.brokers._model.broker_usecase import BrokerAsyncUsecase
+from propan.brokers._model.broker_usecase import BrokerAsyncUsecase, BrokerSyncUsecase
 from propan.brokers._model.routing import BrokerRouter
 from propan.brokers._model.schemas import PropanMessage, Queue
 
 __all__ = (
     "Queue",
     "BrokerAsyncUsecase",
+    "BrokerSyncUsecase",
     "PropanMessage",
     "BrokerRouter",
 )

propan/brokers/_model/broker_usecase.py

@@ -2,7 +2,7 @@
 import logging
 import warnings
 from abc import ABC, abstractmethod
-from contextlib import AsyncExitStack
+from contextlib import AsyncExitStack, ExitStack
 from functools import wraps
 from itertools import chain
 from types import TracebackType
@@ -24,7 +24,6 @@
 from fast_depends._compat import PYDANTIC_V2
 from fast_depends.core import CallModel, build_call_model
 from fast_depends.dependencies import Depends
-from fast_depends.utils import args_to_kwargs
 from typing_extensions import Self, TypeAlias, TypeVar
 
 from propan.brokers._model.routing import BrokerRouter
@@ -150,13 +149,19 @@ def __init__(
 
     def _resolve_connection_kwargs(self, *args: Any, **kwargs: AnyDict) -> AnyDict:
         arguments = get_function_positional_arguments(self.__init__)  # type: ignore
-        init_kwargs = args_to_kwargs(
-            arguments,
-            *self._connection_args,
+
+        init_kwargs = {
             **self._connection_kwargs,
-        )
-        connect_kwargs = args_to_kwargs(arguments, *args, **kwargs)
-        return {**init_kwargs, **connect_kwargs}
+            **dict(zip(arguments, self._connection_args)),
+        }
+
+        connect_kwargs = {
+            **kwargs,
+            **dict(zip(arguments, args)),
+        }
+
+        final_kwargs = {**init_kwargs, **connect_kwargs}
+        return final_kwargs
 
     @staticmethod
     def _decode_message(message: PropanMessage[Any]) -> DecodedMessage:
@@ -866,6 +871,20 @@ def log_wrapper(message: PropanMessage[MsgType]) -> T_HandlerReturn:
 
         return log_wrapper
 
+    def _wrap_middleware(
+        self,
+        func: Callable[[MsgType], Union[T_HandlerReturn, Awaitable[T_HandlerReturn]]],
+    ) -> Callable[[MsgType], Union[T_HandlerReturn, Awaitable[T_HandlerReturn]]]:
+        @wraps(func)
+        def middleware_wrapper(message: MsgType) -> T_HandlerReturn:
+            with ExitStack() as stack:
+                for m in self.middlewares:
+                    stack.enter_context(m(message))
+
+                return func(message)
+
+        return middleware_wrapper
+
 
 def extend_dependencies(extra: Sequence[CallModel]) -> Callable[[CallModel], CallModel]:
     def dependant_wrapper(dependant: CallModel) -> CallModel:

propan/brokers/push_back_watcher.py

@@ -134,3 +134,30 @@ async def __aexit__(
 
         else:
             await call_or_await(self.on_error, self.message, **self.extra_args)
+
+    def __enter__(self) -> None:
+        self.watcher.add(self.message.message_id)
+
+    def __exit__(
+        self,
+        exc_type: Optional[Type[BaseException]],
+        exc_val: Optional[BaseException],
+        exc_tb: Optional[TracebackType],
+    ) -> None:
+        if not exc_type:
+            if callable(self.on_success):
+                try:
+                    self.on_success(self.message, **self.extra_args)
+                except TypeError as e:
+                    print("on_success type error", e)
+            self.watcher.remove(self.message.message_id)
+
+        elif isinstance(exc_val, SkipMessage):
+            self.watcher.remove(self.message.message_id)
+
+        elif self.watcher.is_max(self.message.message_id):
+            self.on_max(self.message, **self.extra_args)
+            self.watcher.remove(self.message.message_id)
+
+        else:
+            self.on_error(self.message, **self.extra_args)

propan/brokers/rabbit/__init__.py

@@ -1,6 +1,7 @@
 from propan.brokers.rabbit.rabbit_broker import RabbitBroker, RabbitMessage
 from propan.brokers.rabbit.routing import RabbitRouter
 from propan.brokers.rabbit.schemas import ExchangeType, RabbitExchange, RabbitQueue
+from propan.brokers.rabbit.rabbit_broker_sync import RabbitSyncBroker
 
 __all__ = (
     "RabbitBroker",
@@ -9,4 +10,5 @@
     "RabbitExchange",
     "ExchangeType",
     "RabbitMessage",
+    "RabbitSyncBroker",
 )

propan/brokers/rabbit/rabbit_broker.py

@@ -302,7 +302,7 @@ async def _parse_message(
             body=message.body,
             headers=message.headers,
             reply_to=message.reply_to or "",
-            message_id=message.message_id,
+            message_id=message.message_id or str(uuid4()),
             content_type=message.content_type or "",
             raw_message=message,
         )

propan/brokers/rabbit/rabbit_broker_sync.py

@@ -1,7 +1,16 @@
 from functools import wraps
 from threading import Event
 from types import TracebackType
-from typing import Any, Callable, List, Optional, Sequence, Tuple, Type, Union
+from typing import (
+    Any,
+    Callable,
+    List,
+    Optional,
+    Sequence,
+    Tuple,
+    Type,
+    Union,
+)
 from uuid import uuid4
 
 import pika
@@ -29,6 +38,7 @@
 from propan.types import AnyDict, DecodedMessage, SendableMessage
 from propan.utils import context
 
+
 PIKA_RAW_MESSAGE: TypeAlias = Tuple[
     blocking_connection.BlockingChannel,
     spec.Basic.Deliver,
@@ -71,13 +81,14 @@ def __init__(
 
         self._channel = None
 
-        self.__max_queue_len = 4
-        self.__max_exchange_len = 4
+        self._max_queue_len = 4
+        self._max_exchange_len = 4
         self._queues = {}
         self._exchanges = {}
 
     def _connect(self, **kwargs: Any) -> blocking_connection.BlockingConnection:
-        connection = pika.BlockingConnection()
+        # TODO: use all kwargs here
+        connection = pika.BlockingConnection(pika.URLParameters(kwargs["url"]))
 
         if self._channel is None:
             max_consumers = self._max_consumers
@@ -134,7 +145,7 @@ def handle(
         *,
         dependencies: Sequence[Depends] = (),
         description: str = "",
-        **original_kwargs: Any,
+        **original_kwargs: AnyDict,
     ) -> Callable[
         [HandlerCallable[T_HandlerReturn]],
         Callable[[Any, bool], T_HandlerReturn],
@@ -186,21 +197,22 @@ def _process_message(
             watcher = NotPushBackWatcher()
 
         @wraps(func)
-        def wrapper(message):
-            channel: blocking_connection.BlockingChannel
-            method_frame: spec.Basic.Deliver
-            header_frame: spec.BasicProperties
+        def wrapper(message: PikaMessage):
             channel, method_frame, header_frame, _ = message.raw_message
 
             context = WatcherContext(
                 watcher,
-                message.message_id,
-                on_success=lambda: channel.basic_ack(method_frame.delivery_tag),
-                on_error=lambda: channel.basic_nack(
-                    method_frame.delivery_tag, requeue=True
+                message,
+                on_success=lambda msg: channel.basic_ack(
+                    method_frame.delivery_tag
+                ),
+                on_error=lambda msg: channel.basic_nack(
+                    method_frame.delivery_tag,
+                    requeue=True,
                 ),
-                on_max=lambda: channel.basic_reject(
-                    method_frame.delivery_tag, requeue=False
+                on_max=lambda msg: channel.basic_reject(
+                    method_frame.delivery_tag,
+                    requeue=False,
                 ),
             )
 
@@ -245,7 +257,7 @@ def publish(
         type_: Optional[str] = None,
         user_id: Optional[str] = None,
         app_id: Optional[str] = None,
-    ) -> DecodedMessage | None:
+    ) -> Optional[DecodedMessage]:
         if self._channel is None:
             raise ValueError("RabbitBroker channel not started yet")
 
@@ -254,6 +266,7 @@ def publish(
         message, content_type = super()._encode_message(message)
 
         response_event: Optional[Event] = None
+        response_msg: Optional[DecodedMessage] = None
         if callback is True:
             if reply_to is not None:
                 raise WRONG_PUBLISH_ARGS
@@ -267,7 +280,9 @@ def handle_response(
                 header_frame: spec.BasicProperties,
                 body: bytes,
             ):
-                # TODO: return message
+                nonlocal response_msg
+                msg = self._parse_message((channel, method_frame, header_frame, body))
+                response_msg = self._decode_message(msg)
                 response_event.set()
 
             response_consumer_tag = self._channel.basic_consume(
@@ -300,8 +315,11 @@ def handle_response(
 
         if response_event is not None:
             self._channel._process_data_events(callback_timeout)
-            response_event.wait()
-            self._channel.basic_cancel(response_consumer_tag)
+            try:
+                response_event.wait()
+                return response_msg
+            finally:
+                self._channel.basic_cancel(response_consumer_tag)
 
     def _init_handler(self, handler: Handler) -> None:
         self.declare_queue(handler.queue)
@@ -376,7 +394,8 @@ def pika_handler_func_wrapper(
         reraise_exc: bool = False,
     ):
         return func(
-            (channel, method_frame, header_frame, body), reraise_exc=reraise_exc
+            (channel, method_frame, header_frame, body),
+            reraise_exc=reraise_exc,
         )
 
     return pika_handler_func_wrapper

propan/brokers/sqs/sqs_broker.py

@@ -407,7 +407,8 @@ def _get_log_context(
 
 
 async def delete_message(
-    message: SQSMessage, connection: Optional[AioBaseClient]
+    message: SQSMessage,
+    connection: Optional[AioBaseClient],
 ) -> None:
     if connection:
         await connection.delete_message(

propan/test/__init__.py

@@ -2,8 +2,10 @@
 
 try:
     from propan.test.rabbit import TestRabbitBroker
+    from propan.test.rabbit_sync import TestRabbitSyncBroker
 except Exception:
     TestRabbitBroker = about.INSTALL_RABBIT
+    TestRabbitSyncBroker = about.INSTALL_RABBIT
 
 try:
     from propan.test.redis import TestRedisBroker
@@ -26,7 +28,7 @@
     TestSQSBroker = about.INSTALL_SQS
 
 assert any(
-    (TestRabbitBroker, TestRedisBroker, TestKafkaBroker, TestNatsBroker, TestSQSBroker)
+    (TestRabbitBroker, TestRedisBroker, TestKafkaBroker, TestNatsBroker, TestSQSBroker, TestRabbitSyncBroker)
 ), about.INSTALL_MESSAGE
 
 __all__ = (
@@ -35,4 +37,5 @@
     "TestKafkaBroker",
     "TestNatsBroker",
     "TestSQSBroker",
+    "TestRabbitSyncBroker",
 )