improve binding/unbinding operarations

Author: DanielePalaia

rabbitmq_amqp_python_client/__init__.py

@@ -13,6 +13,7 @@
 )
 from .environment import Environment
 from .exceptions import (
+    AmqpValidationException,
     ArgumentOutOfRangeException,
     ValidationCodeException,
 )
@@ -73,4 +74,5 @@
     "OffsetSpecification",
     "OutcomeState",
     "Environment",
+    "AmqpValidationException",
 ]

rabbitmq_amqp_python_client/address_helper.py

@@ -1,25 +1,30 @@
 from .entities import BindingSpecification
 from .qpid.proton._message import Message
 
+from typing import Optional
+
 
 def _is_unreserved(char: str) -> bool:
     # According to RFC 3986, unreserved characters are A-Z, a-z, 0-9, '-', '.', '_', and '~'
     return char.isalnum() or char in "-._~"
 
 
-def encode_path_segment(input_string: str) -> str:
+def encode_path_segment(input_string: Optional[str]) -> str:
     encoded = []
 
     # Iterate over each character in the input string
-    for char in input_string:
-        # Check if the character is an unreserved character
-        if _is_unreserved(char):
-            encoded.append(char)  # Append as is
-        else:
-            # Encode character to %HH format
-            encoded.append(f"%{ord(char):02X}")
+    if input_string is not None:
+        for char in input_string:
+            # Check if the character is an unreserved character
+            if _is_unreserved(char):
+                encoded.append(char)  # Append as is
+            else:
+                # Encode character to %HH format
+                encoded.append(f"%{ord(char):02X}")
+
+        return "".join(encoded)
 
-    return "".join(encoded)
+    return ""
 
 
 class AddressHelper:
@@ -60,6 +65,11 @@ def path_address() -> str:
     def binding_path_with_exchange_queue(
         bind_specification: BindingSpecification,
     ) -> str:
+        if bind_specification.binding_key is not None:
+            key = ";key=" + encode_path_segment(bind_specification.binding_key)
+        else:
+            key = ";key="
+
         binding_path_wth_exchange_queue_key = (
             "/bindings"
             + "/"
@@ -68,11 +78,28 @@ def binding_path_with_exchange_queue(
             + ";"
             + "dstq="
             + encode_path_segment(bind_specification.destination_queue)
+            + key
+            + ";args="
+        )
+        return binding_path_wth_exchange_queue_key
+
+    @staticmethod
+    def binding_path_with_exchange_exchange(
+        bind_specification: BindingSpecification,
+    ) -> str:
+        binding_path_wth_exchange_exchange_key = (
+            "/bindings"
+            + "/"
+            + "src="
+            + encode_path_segment(bind_specification.source_exchange)
+            + ";"
+            + "dstq="
+            + encode_path_segment(bind_specification.destination_exchange)
             + ";key="
             + encode_path_segment(bind_specification.binding_key)
             + ";args="
         )
-        return binding_path_wth_exchange_queue_key
+        return binding_path_wth_exchange_exchange_key
 
     @staticmethod
     def message_to_address_helper(message: Message, address: str) -> Message:

rabbitmq_amqp_python_client/entities.py

@@ -43,8 +43,9 @@ class OffsetSpecification(Enum):
 @dataclass
 class BindingSpecification:
     source_exchange: str
-    destination_queue: str
-    binding_key: str
+    binding_key: Optional[str] = None
+    destination_exchange: Optional[str] = None
+    destination_queue: Optional[str] = None
 
 
 class StreamOptions:

rabbitmq_amqp_python_client/exceptions.py

@@ -14,3 +14,12 @@ def __init__(self, msg: str):
 
     def __str__(self) -> str:
         return repr(self.msg)
+
+
+class AmqpValidationException(BaseException):
+    # Constructor or Initializer
+    def __init__(self, msg: str):
+        self.msg = msg
+
+    def __str__(self) -> str:
+        return repr(self.msg)

rabbitmq_amqp_python_client/management.py

@@ -9,7 +9,10 @@
     ExchangeSpecification,
     QueueInfo,
 )
-from .exceptions import ValidationCodeException
+from .exceptions import (
+    AmqpValidationException,
+    ValidationCodeException,
+)
 from .options import ReceiverOption, SenderOption
 from .qpid.proton._message import Message
 from .qpid.proton.utils import (
@@ -303,10 +306,19 @@ def _validate_reponse_code(
 
     def bind(self, bind_specification: BindingSpecification) -> str:
         logger.debug("Bind Operation called")
+        self._validate_binding(bind_specification)
+
         body = {}
-        body["binding_key"] = bind_specification.binding_key
+        if bind_specification.binding_key is not None:
+            body["binding_key"] = bind_specification.binding_key
+        else:
+            body["binding_key"] = ""
         body["source"] = bind_specification.source_exchange
-        body["destination_queue"] = bind_specification.destination_queue
+        if bind_specification.destination_queue is not None:
+            body["destination_queue"] = bind_specification.destination_queue
+        elif bind_specification.destination_exchange is not None:
+            body["destination_exchange"] = bind_specification.destination_exchange
+
         body["arguments"] = {}  # type: ignore
 
         path = AddressHelper.path_address()
@@ -320,16 +332,53 @@ def bind(self, bind_specification: BindingSpecification) -> str:
             ],
         )
 
-        binding_path_with_queue = AddressHelper.binding_path_with_exchange_queue(
-            bind_specification
-        )
-        return binding_path_with_queue
+        if bind_specification.destination_queue is not None:
+            binding_path = AddressHelper.binding_path_with_exchange_queue(
+                bind_specification
+            )
+        elif bind_specification.destination_exchange is not None:
+            binding_path = AddressHelper.binding_path_with_exchange_exchange(
+                bind_specification
+            )
 
-    def unbind(self, binding_exchange_queue_path: str) -> None:
+        return binding_path
+
+    def _validate_binding(self, bind_specification: BindingSpecification) -> None:
+        if (
+            bind_specification.destination_queue is not None
+            and bind_specification.destination_exchange is not None
+        ):
+            raise AmqpValidationException(
+                "just one of destination_queue and destination_exchange of BindingSpecification must be set "
+                "for a binding operation"
+            )
+
+        if (
+            bind_specification.destination_queue is None
+            and bind_specification.destination_exchange is None
+        ):
+            raise AmqpValidationException(
+                "at least one of destination_queue and destination_exchange of BindingSpecification must be set "
+                "for a binding operation"
+            )
+
+    def unbind(self, bind_specification: Union[BindingSpecification, str]) -> None:
         logger.debug("UnBind Operation called")
+        if isinstance(bind_specification, str):
+            binding_name = bind_specification
+        else:
+            self._validate_binding(bind_specification)
+            if bind_specification.destination_queue is not None:
+                binding_name = AddressHelper.binding_path_with_exchange_queue(
+                    bind_specification
+                )
+            elif bind_specification.destination_exchange is not None:
+                binding_name = AddressHelper.binding_path_with_exchange_exchange(
+                    bind_specification
+                )
         self.request(
             None,
-            binding_exchange_queue_path,
+            binding_name,
             CommonValues.command_delete.value,
             [
                 CommonValues.response_code_204.value,

tests/test_management.py

@@ -1,6 +1,7 @@
 from datetime import timedelta
 
 from rabbitmq_amqp_python_client import (
+    AmqpValidationException,
     BindingSpecification,
     ClassicQueueSpecification,
     ExchangeSpecification,
@@ -98,6 +99,126 @@ def test_bind_exchange_to_queue(management: Management) -> None:
     management.unbind(binding_exchange_queue_path)
 
 
+def test_bind_no_destination(management: Management) -> None:
+
+    exchange_name = "test-bind-exchange-to-queue-exchange"
+    queue_name = "test-bind-exchange-to-queue-queue"
+    routing_key = "routing-key"
+    raised = False
+
+    management.declare_exchange(ExchangeSpecification(name=exchange_name))
+
+    management.declare_queue(QuorumQueueSpecification(name=queue_name))
+
+    try:
+        management.bind(
+            BindingSpecification(
+                source_exchange=exchange_name,
+                binding_key=routing_key,
+            )
+        )
+    except AmqpValidationException:
+        raised = True
+
+    assert raised is True
+
+    management.delete_exchange(exchange_name)
+
+    management.delete_queue(queue_name)
+
+
+def test_bind_exchange_to_queue_without_key(management: Management) -> None:
+
+    exchange_name = "test-bind-exchange-to-queue-exchange"
+    queue_name = "test-bind-exchange-to-queue-queue"
+
+    management.declare_exchange(ExchangeSpecification(name=exchange_name))
+
+    management.declare_queue(QuorumQueueSpecification(name=queue_name))
+
+    binding_exchange_queue_path = management.bind(
+        BindingSpecification(
+            source_exchange=exchange_name,
+            destination_queue=queue_name,
+        )
+    )
+
+    assert (
+        binding_exchange_queue_path
+        == "/bindings/src=" + exchange_name + ";dstq=" + queue_name + ";key=" + ";args="
+    )
+
+    management.unbind(binding_exchange_queue_path)
+
+    management.delete_exchange(exchange_name)
+
+    management.delete_queue(queue_name)
+
+
+def test_bind_unbind_by_binding_spec(management: Management) -> None:
+
+    exchange_name = "test-bind-exchange-to-queue-exchange"
+    queue_name = "test-bind-exchange-to-queue-queue"
+
+    management.declare_exchange(ExchangeSpecification(name=exchange_name))
+
+    management.declare_queue(QuorumQueueSpecification(name=queue_name))
+
+    management.bind(
+        BindingSpecification(
+            source_exchange=exchange_name,
+            destination_queue=queue_name,
+        )
+    )
+
+    management.unbind(
+        BindingSpecification(
+            source_exchange=exchange_name,
+            destination_queue=queue_name,
+        )
+    )
+
+    management.delete_exchange(exchange_name)
+
+    management.delete_queue(queue_name)
+
+
+def test_bind_exchange_to_exchange(management: Management) -> None:
+
+    source_exchange_name = "source_exchange"
+    destination_exchange_name = "destination_exchange"
+    routing_key = "routing-key"
+
+    management.declare_exchange(ExchangeSpecification(name=source_exchange_name))
+
+    management.declare_exchange(ExchangeSpecification(name=destination_exchange_name))
+
+    binding_exchange_exchange_path = management.bind(
+        BindingSpecification(
+            source_exchange=source_exchange_name,
+            destination_exchange=destination_exchange_name,
+            binding_key=routing_key,
+        )
+    )
+
+    assert (
+        binding_exchange_exchange_path
+        == "/bindings/src="
+        + source_exchange_name
+        + ";dstq="
+        + destination_exchange_name
+        + ";key="
+        + routing_key
+        + ";args="
+    )
+
+    management.unbind(binding_exchange_exchange_path)
+
+    management.delete_exchange(source_exchange_name)
+
+    management.delete_exchange(destination_exchange_name)
+
+
 def test_queue_info_with_validations(management: Management) -> None:
 
     queue_name = "test_queue_info_with_validation"