improving url helper

Author: DanielePalaia

rabbitmq_amqp_python_client/address_helper.py

@@ -1,17 +1,57 @@
 from .entities import BindingSpecification
 
 
+def binding_path_with_exchange_queue(bind_specification: BindingSpecification) -> str:
+    binding_path_wth_exchange_queue_key = (
+        "/bindings"
+        + "/"
+        + "src="
+        + bind_specification.source_exchange
+        + ";"
+        + "dstq="
+        + bind_specification.destination_queue
+        + ";key="
+        + bind_specification.binding_key
+        + ";args="
+    )
+    return binding_path_wth_exchange_queue_key
+
+
+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(queue_name: str) -> str:
+    path = "/queues/" + encode_path_segment(queue_name) + "/messages"
 
     return path
 
@@ -28,17 +68,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