implementing Declare Exchange/Queue

Author: DanielePalaia

examples/getting_started/main.py

@@ -3,11 +3,14 @@
 from rabbitmq_amqp_python_client import (
     Connection,
     ExchangeSpecification,
+    QueueSpecification,
+    QueueType,
 )
 
 
 def main():
     exchange_name = "example-exchange"
+    queue_name = "example-queue"
     connection = Connection("amqp://guest:guest@localhost:5672/")
 
     connection.dial()
@@ -18,12 +21,9 @@ def main():
         ExchangeSpecification(name=exchange_name, arguments={})
     )
 
-    """ 
-    queue_info = management.declare_queue(QueueSpecification{
-		name:      queue_name,       
-		queue_type: QueueType{type: Quorum},
-	})
-    """
+    queue_info = management.declare_queue(
+        QueueSpecification(name=queue_name, queue_type=QueueType.quorum, arguments={})
+    )
 
     """
     #management.bind(BindingSpecification{
@@ -67,7 +67,9 @@ def main():
     management.delete_exchange(exchange_info.name)
     """
 
-    # connection.close()
+    management.close()
+
+    connection.close()
 
 
 if __name__ == "__main__":

rabbitmq_amqp_python_client/__init__.py

@@ -1,7 +1,11 @@
 from importlib import metadata
 
+from .common import QueueType
 from .connection import Connection
-from .entities import ExchangeSpecification
+from .entities import (
+    ExchangeSpecification,
+    QueueSpecification,
+)
 
 try:
     __version__ = metadata.version(__package__)
@@ -15,4 +19,6 @@
 __all__ = [
     "Connection",
     "ExchangeSpecification",
+    "QueueSpecification",
+    "QueueType",
 ]

rabbitmq_amqp_python_client/address_helper.py

@@ -2,3 +2,9 @@ def exchange_address(name: str) -> str:
     path = "/exchanges/" + name
 
     return path
+
+
+def queue_address(name: str) -> str:
+    path = "/queues/" + name
+
+    return path

rabbitmq_amqp_python_client/common.py

@@ -20,7 +20,13 @@ class CommonValues(enum.Enum):
     bindings = "bindings"
 
 
-class ExchangeTypes(enum.Enum):
+class ExchangeType(enum.Enum):
     direct = "direct"
     topic = "topic"
     fanout = "fanout"
+
+
+class QueueType(enum.Enum):
+    quorum = "quorum"
+    classic = "classic"
+    stream = "stream"

rabbitmq_amqp_python_client/entities.py

@@ -1,18 +1,26 @@
-import enum
-from collections import defaultdict
 from dataclasses import dataclass
+from typing import Optional
 
-
-class ExchangeType(enum.Enum):
-    direct = "direct"
-    topic = "topic"
-    fanout = "fanout"
+from .common import ExchangeType, QueueType
 
 
 @dataclass
 class ExchangeSpecification:
     name: str
     arguments: dict
+    exchange_type: ExchangeType = ExchangeType.direct
+    is_auto_delete: bool = False
+    is_durable: bool = True
+
+
+@dataclass
+class QueueSpecification:
+    name: str
+    arguments: dict
+    queue_type: QueueType = QueueType.quorum
+    dead_letter_routing_key: str = ""
+    is_exclusive: Optional[bool] = None
+    max_len_bytes: Optional[int] = None
+    dead_letter_exchange: str = ""
     is_auto_delete: bool = False
     is_durable: bool = True
-    exchange_type: ExchangeType = ExchangeType.direct

rabbitmq_amqp_python_client/management.py

@@ -8,13 +8,16 @@
     BlockingSender,
 )
 
-from .address_helper import exchange_address
+from .address_helper import exchange_address, queue_address
 from .common import CommonValues
 from .configuration_options import (
     ReceiverOption,
     SenderOption,
 )
-from .entities import ExchangeSpecification
+from .entities import (
+    ExchangeSpecification,
+    QueueSpecification,
+)
 
 
 class Management:
@@ -85,9 +88,6 @@ def _request(
 
         # TO_COMPLETE HERE
 
-    # TODO
-    # def declare_queue(self, name:str):
-
     # TODO
     # def delete_queue(self, name:str):
 
@@ -114,6 +114,32 @@ def declare_exchange(self, exchange_specification: ExchangeSpecification):
             ],
         )
 
+    def declare_queue(self, queue_specification: QueueSpecification):
+        body = {}
+        body["auto_delete"] = queue_specification.is_auto_delete
+        body["durable"] = queue_specification.is_durable
+        body["arguments"] = {
+            "x-queue-type": queue_specification.queue_type.value,
+            "x-dead-letter-exchange": queue_specification.dead_letter_exchange,
+            "x-dead-letter-routing-key": queue_specification.dead_letter_routing_key,
+            "max-length-bytes": queue_specification.max_len_bytes,
+        }
+
+        path = queue_address(queue_specification.name)
+
+        print(path)
+
+        self.request(
+            body,
+            path,
+            CommonValues.command_put.value,
+            [
+                CommonValues.response_code_201.value,
+                CommonValues.response_code_204.value,
+                CommonValues.response_code_409.value,
+            ],
+        )
+
     # TODO
     # def delete_exchange(self, name:str):