adding headers and custom ExchangeTypes

Author: DanielePalaia

rabbitmq_amqp_python_client/common.py

@@ -24,6 +24,7 @@ class ExchangeType(enum.Enum):
     direct = "direct"
     topic = "topic"
     fanout = "fanout"
+    headers = "headers"
 
 
 class QueueType(enum.Enum):

rabbitmq_amqp_python_client/entities.py

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

rabbitmq_amqp_python_client/management.py

@@ -3,7 +3,7 @@
 from typing import Any, Optional, Union
 
 from .address_helper import AddressHelper
-from .common import CommonValues, QueueType
+from .common import CommonValues, ExchangeType, QueueType
 from .entities import (
     ExchangeSpecification,
     ExchangeToExchangeBindingSpecification,
@@ -101,12 +101,15 @@ def declare_exchange(
         self, exchange_specification: ExchangeSpecification
     ) -> ExchangeSpecification:
         logger.debug("declare_exchange operation called")
-        body = {}
+        body: dict[str, Any] = {}
         body["auto_delete"] = exchange_specification.is_auto_delete
         body["durable"] = exchange_specification.is_durable
-        body["type"] = exchange_specification.exchange_type.value  # type: ignore
+        if isinstance(exchange_specification.exchange_type, ExchangeType):
+            body["type"] = exchange_specification.exchange_type.value
+        else:
+            body["type"] = str(exchange_specification.exchange_type)
         body["internal"] = exchange_specification.is_internal
-        body["arguments"] = exchange_specification.arguments  # type: ignore
+        body["arguments"] = exchange_specification.arguments
 
         path = AddressHelper.exchange_address(exchange_specification.name)
 

tests/test_management.py

@@ -29,6 +29,39 @@ def test_declare_delete_exchange(management: Management) -> None:
     management.delete_exchange(exchange_name)
 
 
+def test_declare_delete_exchange_headers(management: Management) -> None:
+
+    exchange_name = "test-exchange"
+
+    exchange_info = management.declare_exchange(
+        ExchangeSpecification(name=exchange_name, exchange_type=ExchangeType.headers)
+    )
+
+    assert exchange_info.name == exchange_name
+
+    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"