Include urlencode, urldecode, quote, unquote implementations, outside of stdlib. (#83)

* urlencode/urldecode implementations, outside of stdlib

* No need to lazy-import json

* Dont lazy import json

Author: lovelydinosaur

.gitignore

@@ -1,6 +1,7 @@
 *.pyc
 .coverage
 .mypy_cache/
+.pytest_cache/
 __pycache__/
 dist/
 venv/

scripts/unasync

@@ -11,6 +11,7 @@ unasync.unasync_files(
         "src/ahttpx/_response.py",
         "src/ahttpx/_request.py",
         "src/ahttpx/_streams.py",
+        "src/ahttpx/_urlencode.py",
         "src/ahttpx/_urlparse.py",
         "src/ahttpx/_urls.py"
     ],

src/ahttpx/__init__.py

@@ -7,6 +7,7 @@
 from ._request import *  # Request
 from ._streams import *  # ByteStream, IterByteStream, Stream
 from ._server import *  # serve_http, serve_tcp
+from ._urlencode import *  # quote, unquote, urldecode, urlencode
 from ._urls import *  # QueryParams, URL
 
 
@@ -38,5 +39,9 @@
     "timeout",
     "Transport",
     "QueryParams",
+    "quote",
+    "unquote",
     "URL",
+    "urldecode",
+    "urlencode",
 ]

src/ahttpx/_content.py

@@ -1,8 +1,9 @@
+import json
 import os
 import typing
-import urllib.parse
 
 from ._streams import Stream, ByteStream, IterByteStream
+from ._urlencode import urldecode, urlencode
 
 __all__ = [
     "Content",
@@ -87,7 +88,7 @@ def __init__(
         if form is None:
             d = {}
         elif isinstance(form, str):
-            d = urllib.parse.parse_qs(form, keep_blank_values=True)
+            d = urldecode(form)
         elif isinstance(form, Form):
             d = form.multi_dict()
         elif isinstance(form, typing.Mapping):
@@ -187,7 +188,7 @@ def __eq__(self, other: typing.Any) -> bool:
         )
 
     def __str__(self) -> str:
-        return urllib.parse.urlencode(self.multi_items())
+        return urlencode(self.multi_dict())
 
     def __repr__(self) -> str:
         return f"<Form {self.multi_dict()!r}>"
@@ -317,8 +318,6 @@ def __init__(self, data: typing.Any) -> None:
         self._data = data
 
     def encode(self) -> tuple[Stream, str]:
-        import json
-
         content = json.dumps(
             self._data,
             ensure_ascii=False,

src/ahttpx/_pool.py

@@ -128,7 +128,7 @@ def description(self) -> str:
 
     # Builtins...
     def __repr__(self) -> str:
-        return f"<ConectionPool [{self.description()}]>"
+        return f"<ConnectionPool [{self.description()}]>"
 
     async def __aenter__(self) -> "ConnectionPool":
         return self

src/ahttpx/_urlencode.py

@@ -0,0 +1,85 @@
+import re
+
+__all__ = ["quote", "unquote", "urldecode", "urlencode"]
+
+
+# Matchs a sequence of one or more '%xx' escapes.
+PERCENT_ENCODED_REGEX = re.compile("(%[A-Fa-f0-9][A-Fa-f0-9])+")
+
+# https://datatracker.ietf.org/doc/html/rfc3986#section-2.3
+SAFE = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-._~"
+
+
+def urlencode(multidict, safe=SAFE):
+    pairs = []
+    for key, values in multidict.items():
+        pairs.extend([(key, value) for value in values])
+
+    safe += "+"
+    pairs = [(k.replace(" ", "+"), v.replace(" ", "+")) for k, v in pairs]
+
+    return "&".join(
+        f"{quote(key, safe)}={quote(val, safe)}"
+        for key, val in pairs
+    )
+
+
+def urldecode(string):
+    parts = [part.partition("=") for part in string.split("&") if part]
+    pairs = [
+        (unquote(key), unquote(val))
+        for key, _, val in parts
+    ]
+
+    pairs = [(k.replace("+", " "), v.replace("+", " ")) for k, v in pairs]
+
+    ret = {}
+    for k, v in pairs:
+        ret.setdefault(k, []).append(v)
+    return ret
+
+
+def quote(string, safe=SAFE):
+    # Fast path if the string is already safe.
+    if not string.strip(safe):
+        return string
+
+    # Replace any characters not in the safe set with '%xx' escape sequences.
+    return "".join([
+        char if char in safe else percent(char)
+        for char in string
+    ])
+
+
+def unquote(string):
+    # Fast path if the string is not quoted.
+    if '%' not in string:
+        return string
+
+    # Unquote.
+    parts = []
+    current_position = 0
+    for match in re.finditer(PERCENT_ENCODED_REGEX, string):
+        start_position, end_position = match.start(), match.end()
+        matched_text = match.group(0)
+        # Include any text up to the '%xx' escape sequence.
+        if start_position != current_position:
+            leading_text = string[current_position:start_position]
+            parts.append(leading_text)
+
+        # Decode the '%xx' escape sequence.
+        hex = matched_text.replace('%', '')
+        decoded = bytes.fromhex(hex).decode('utf-8')
+        parts.append(decoded)
+        current_position = end_position
+
+    # Include any text after the final '%xx' escape sequence.
+    if current_position != len(string):
+        trailing_text = string[current_position:]
+        parts.append(trailing_text)
+
+    return "".join(parts)
+
+
+def percent(c):
+    return ''.join(f"%{b:02X}" for b in c.encode("utf-8"))

src/ahttpx/_urls.py

@@ -1,11 +1,11 @@
-from  __future__ import annotations
+from __future__ import annotations
 
 import typing
-from urllib.parse import parse_qs, unquote, urlencode
 
 from ._urlparse import urlparse
+from ._urlencode import unquote, urldecode, urlencode
 
-__all__ = ["URL", "QueryParams"]
+__all__ = ["QueryParams", "URL"]
 
 
 class URL:
@@ -70,7 +70,7 @@ class URL:
       themselves.
     """
 
-    def __init__(self, url: URL | str = "", **kwargs: typing.Any) -> None:
+    def __init__(self, url: "URL" | str = "", **kwargs: typing.Any) -> None:
         if kwargs:
             allowed = {
                 "scheme": str,
@@ -233,7 +233,7 @@ def query(self) -> bytes:
         return query.encode("ascii")
 
     @property
-    def params(self) -> QueryParams:
+    def params(self) -> "QueryParams":
         """
         The URL query parameters, neatly parsed and packaged into an immutable
         multidict representation.
@@ -358,7 +358,7 @@ def __init__(
         if params is None:
             d = {}
         elif isinstance(params, str):
-            d = parse_qs(params, keep_blank_values=True)
+            d = urldecode(params)
         elif isinstance(params, QueryParams):
             d = params.multi_dict()
         elif isinstance(params, dict):
@@ -454,7 +454,7 @@ def get_list(self, key: str) -> list[str]:
         """
         return list(self._dict.get(key, []))
 
-    def set(self, key: str, value: str) -> "QueryParams":
+    def copy_set(self, key: str, value: str) -> "QueryParams":
         """
         Return a new QueryParams instance, setting the value of a key.
 
@@ -469,7 +469,7 @@ def set(self, key: str, value: str) -> "QueryParams":
         q._dict[key] = [value]
         return q
 
-    def append(self, key: str, value: str) -> "QueryParams":
+    def copy_append(self, key: str, value: str) -> "QueryParams":
         """
         Return a new QueryParams instance, setting or appending the value of a key.
 
@@ -484,7 +484,7 @@ def append(self, key: str, value: str) -> "QueryParams":
         q._dict[key] = q.get_list(key) + [value]
         return q
 
-    def remove(self, key: str) -> QueryParams:
+    def copy_remove(self, key: str) -> QueryParams:
         """
         Return a new QueryParams instance, removing the value of a key.
 
@@ -499,7 +499,7 @@ def remove(self, key: str) -> QueryParams:
         q._dict.pop(str(key), None)
         return q
 
-    def copy_with(
+    def copy_update(
         self,
         params: (
             "QueryParams" | dict[str, str | list[str]] | list[tuple[str, str]] | None
@@ -511,11 +511,11 @@ def copy_with(
         Usage:
 
         q = httpx.QueryParams("a=123")
-        q = q.copy_with({"b": "456"})
+        q = q.copy_update({"b": "456"})
         assert q == httpx.QueryParams("a=123&b=456")
 
         q = httpx.QueryParams("a=123")
-        q = q.copy_with({"a": "456", "b": "789"})
+        q = q.copy_update({"a": "456", "b": "789"})
         assert q == httpx.QueryParams("a=456&b=789")
         """
         q = QueryParams(params)
@@ -546,7 +546,7 @@ def __eq__(self, other: typing.Any) -> bool:
         return sorted(self.multi_items()) == sorted(other.multi_items())
 
     def __str__(self) -> str:
-        return urlencode(self.multi_items())
+        return urlencode(self.multi_dict())
 
     def __repr__(self) -> str:
         return f"<QueryParams {str(self)!r}>"

src/httpx/__init__.py

@@ -7,6 +7,7 @@
 from ._request import *  # Request
 from ._streams import *  # ByteStream, IterByteStream, Stream
 from ._server import *  # serve_http, serve_tcp
+from ._urlencode import *  # quote, unquote, urldecode, urlencode
 from ._urls import *  # QueryParams, URL
 
 
@@ -38,5 +39,9 @@
     "timeout",
     "Transport",
     "QueryParams",
+    "quote",
+    "unquote",
     "URL",
+    "urldecode",
+    "urlencode",
 ]

src/httpx/_content.py

@@ -1,8 +1,9 @@
+import json
 import os
 import typing
-import urllib.parse
 
 from ._streams import Stream, ByteStream, IterByteStream
+from ._urlencode import urldecode, urlencode
 
 __all__ = [
     "Content",
@@ -87,7 +88,7 @@ def __init__(
         if form is None:
             d = {}
         elif isinstance(form, str):
-            d = urllib.parse.parse_qs(form, keep_blank_values=True)
+            d = urldecode(form)
         elif isinstance(form, Form):
             d = form.multi_dict()
         elif isinstance(form, typing.Mapping):
@@ -187,7 +188,7 @@ def __eq__(self, other: typing.Any) -> bool:
         )
 
     def __str__(self) -> str:
-        return urllib.parse.urlencode(self.multi_items())
+        return urlencode(self.multi_dict())
 
     def __repr__(self) -> str:
         return f"<Form {self.multi_dict()!r}>"
@@ -317,8 +318,6 @@ def __init__(self, data: typing.Any) -> None:
         self._data = data
 
     def encode(self) -> tuple[Stream, str]:
-        import json
-
         content = json.dumps(
             self._data,
             ensure_ascii=False,