improving url helper

Author: DanielePalaia

rabbitmq_amqp_python_client/address_helper.py

@@ -1,23 +1,41 @@
 from .entities import BindingSpecification
 
 
+def encode_path_segment(input: str) -> str:
+    encoded_input = ""
+    for character in input:
+        if is_unreserved(character):
+            encoded_input += character
+
+        else:
+            # encoded_input += "%%%02X" % character
+            encoded_input += hex((ord(character)))
+
+    return encoded_input
+
+
 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
 
@@ -28,17 +46,29 @@ def path_address() -> str:
     return path
 
 
+def is_unreserved(input: str) -> bool:
+    return (
+        (input >= "A" and input <= "Z")
+        or (input >= "a" and input <= "z")
+        or (input >= "0" and input <= "9")
+        or input == "-"
+        or input == "."
+        or input == "_"
+        or input == "~"
+    )
+
+
 def binding_path_with_exchange_queue(bind_specification: BindingSpecification) -> str:
     binding_path_wth_exchange_queue_key = (
         "/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

tests/test_address_helper.py

@@ -1,18 +1,17 @@
-"""
 from rabbitmq_amqp_python_client import queue_address
 
+
 def test_encoding_simple() -> None:
     queue = "my_queue"
 
     address = queue_address(queue)
 
     assert address == "/queues/my_queue"
 
+
 def test_encoding_hex() -> None:
     queue = "my_queue>"
 
     address = queue_address(queue)
 
     assert address == "/queues/my_queue%3E"
-
-"""