improving url helper

Author: DanielePalaia

rabbitmq_amqp_python_client/address_helper.py

@@ -1,23 +1,48 @@
 from .entities import BindingSpecification
 
 
+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:
+    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}")
+
+    return "".join(encoded)
+
+
 def exchange_address(exchange_name: str, routing_key: str = "") -> str:
     if routing_key == "":
-        path = "/exchanges/" + exchange_name
+        path = "/exchanges/" + encode_path_segment(exchange_name)
     else:
-        path = "/exchanges/" + exchange_name + "/" + routing_key
+        path = (
+            "/exchanges/"
+            + encode_path_segment(exchange_name)
+            + "/"
+            + encode_path_segment(routing_key)
+        )
 
     return path
 
 
-def queue_address(name: str) -> str:
-    path = "/queues/" + name
+def queue_address(queue_name: str) -> str:
+    path = "/queues/" + encode_path_segment(queue_name)
 
     return path
 
 
-def purge_queue_address(name: str) -> str:
-    path = "/queues/" + name + "/messages"
+def purge_queue_address(queue_name: str) -> str:
+    path = "/queues/" + encode_path_segment(queue_name) + "/messages"
 
     return path
 
@@ -33,12 +58,12 @@ def binding_path_with_exchange_queue(bind_specification: BindingSpecification) -
         "/bindings"
         + "/"
         + "src="
-        + bind_specification.source_exchange
+        + encode_path_segment(bind_specification.source_exchange)
         + ";"
         + "dstq="
-        + bind_specification.destination_queue
+        + encode_path_segment(bind_specification.destination_queue)
         + ";key="
-        + bind_specification.binding_key
+        + encode_path_segment(bind_specification.binding_key)
         + ";args="
     )
     return binding_path_wth_exchange_queue_key

rabbitmq_amqp_python_client/management.py

@@ -230,7 +230,7 @@ def unbind(self, binding_exchange_queue_path: str) -> None:
     # def queue_info(self, queue_name:str):
 
     # TODO
-    def purge_queue(self, queue_name: str):
+    def purge_queue(self, queue_name: str) -> None:
         logger.debug("purge_queue operation called")
         path = purge_queue_address(queue_name)
 

tests/test_address_helper.py

@@ -1,18 +1,28 @@
-"""
-from rabbitmq_amqp_python_client import queue_address
+from rabbitmq_amqp_python_client import (
+    exchange_address,
+    queue_address,
+)
 
-def test_encoding_simple() -> None:
+
+def test_encoding_queue_simple() -> None:
     queue = "my_queue"
 
     address = queue_address(queue)
 
     assert address == "/queues/my_queue"
 
-def test_encoding_hex() -> None:
+
+def test_encoding_queue_hex() -> None:
     queue = "my_queue>"
 
     address = queue_address(queue)
 
     assert address == "/queues/my_queue%3E"
 
-"""
+
+def test_encoding_exchange_hex() -> None:
+    queue = "my_exchange/()"
+
+    address = exchange_address(queue)
+
+    assert address == "/exchanges/my_exchange%2F%28%29"