adding a class for CustomExchangeSpec

Author: DanielePalaia

rabbitmq_amqp_python_client/__init__.py

@@ -6,6 +6,7 @@
 from .connection import Connection
 from .consumer import Consumer
 from .entities import (
+    ExchangeCustomSpecification,
     ExchangeSpecification,
     ExchangeToExchangeBindingSpecification,
     ExchangeToQueueBindingSpecification,
@@ -75,4 +76,5 @@
     "OffsetSpecification",
     "OutcomeState",
     "Environment",
+    "ExchangeCustomSpecification",
 ]

rabbitmq_amqp_python_client/entities.py

@@ -14,7 +14,17 @@
 class ExchangeSpecification:
     name: str
     arguments: dict[str, str] = field(default_factory=dict)
-    exchange_type: Union[ExchangeType, str] = ExchangeType.direct
+    exchange_type: ExchangeType = ExchangeType.direct
+    is_auto_delete: bool = False
+    is_internal: bool = False
+    is_durable: bool = True
+
+
+@dataclass
+class ExchangeCustomSpecification:
+    name: str
+    exchange_type: str
+    arguments: dict[str, str] = field(default_factory=dict)
     is_auto_delete: bool = False
     is_internal: bool = False
     is_durable: bool = True

rabbitmq_amqp_python_client/management.py

@@ -3,8 +3,9 @@
 from typing import Any, Optional, Union
 
 from .address_helper import AddressHelper
-from .common import CommonValues, ExchangeType, QueueType
+from .common import CommonValues, QueueType
 from .entities import (
+    ExchangeCustomSpecification,
     ExchangeSpecification,
     ExchangeToExchangeBindingSpecification,
     ExchangeToQueueBindingSpecification,
@@ -98,16 +99,19 @@ def _request(
         return msg
 
     def declare_exchange(
-        self, exchange_specification: ExchangeSpecification
-    ) -> ExchangeSpecification:
+        self,
+        exchange_specification: Union[
+            ExchangeSpecification, ExchangeCustomSpecification
+        ],
+    ) -> Union[ExchangeSpecification, ExchangeCustomSpecification]:
         logger.debug("declare_exchange operation called")
         body: dict[str, Any] = {}
         body["auto_delete"] = exchange_specification.is_auto_delete
         body["durable"] = exchange_specification.is_durable
-        if isinstance(exchange_specification.exchange_type, ExchangeType):
+        if isinstance(exchange_specification, ExchangeSpecification):
             body["type"] = exchange_specification.exchange_type.value
-        else:
-            body["type"] = str(exchange_specification.exchange_type)
+        elif isinstance(exchange_specification, ExchangeCustomSpecification):
+            body["type"] = exchange_specification.exchange_type
         body["internal"] = exchange_specification.is_internal
         body["arguments"] = exchange_specification.arguments
 

tests/test_management.py

@@ -42,26 +42,6 @@ def test_declare_delete_exchange_headers(management: Management) -> None:
     management.delete_exchange(exchange_name)
 
 
-def test_declare_delete_exchange_custom(management: Management) -> None:
-
-    exchange_name = "test-exchange"
-
-    exchange_arguments = {}
-    exchange_arguments["x-delayed-type"] = "direct"
-
-    exchange_info = management.declare_exchange(
-        ExchangeSpecification(
-            name=exchange_name,
-            exchange_type="x-local-random",
-            arguments=exchange_arguments,
-        )
-    )
-
-    assert exchange_info.name == exchange_name
-
-    management.delete_exchange(exchange_name)
-
-
 def test_declare_delete_exchange_with_args(management: Management) -> None:
 
     exchange_name = "test-exchange-with-args"