⚡ various optimizations (#22)

* ⚡ refactor merge logic for improved readability and performance in utils

* ⚡ optimize encoding logic for improved performance and clarity in encode_utils

* ⚡ optimize decode logic for improved performance and clarity in decode_utils

* ⚡ optimize key handling and object normalization for improved performance and clarity in encode.py

* ⚡ optimize delimiter splitting and list parsing logic for improved performance and clarity in decode.py

* ⚡ optimize proxy caching and dict hashing logic for improved performance and determinism in weak_wrapper

* ⚡ optimize Undefined singleton logic for thread safety and clarity; prevent subclassing and ensure identity preservation

* ⚡ optimize EncodeOptions initialization and equality logic for improved clarity and determinism

* ⚡ optimize DecodeOptions post-init logic for improved determinism and enforce consistency between decode_dot_in_keys and allow_dots

* ⚡ optimize list merging logic in Utils.merge for improved determinism and handling of Undefined values

* ⚡ optimize type checking in list merging logic for improved clarity and consistency in Utils.merge

* ⚡ optimize encode logic for improved determinism and clarity; use UNDEFINED singleton and refine ListFormat.COMMA comparison

* ⚡ optimize decode logic to use UNDEFINED singleton for list initialization

Author: techouse

src/qs_codec/decode.py

@@ -20,7 +20,7 @@
 from .enums.duplicates import Duplicates
 from .enums.sentinel import Sentinel
 from .models.decode_options import DecodeOptions
-from .models.undefined import Undefined
+from .models.undefined import UNDEFINED
 from .utils.decode_utils import DecodeUtils
 from .utils.utils import Utils
 
@@ -168,19 +168,17 @@ def _parse_query_string_values(value: str, options: DecodeOptions) -> t.Dict[str
         raise ValueError("Parameter limit must be a positive integer.")
 
     parts: t.List[str]
-    # Split using either a compiled regex or a literal string delimiter.
+    # Split using either a compiled regex or a literal string delimiter (do it once, then slice if needed).
     if isinstance(options.delimiter, re.Pattern):
-        parts = (
-            re.split(options.delimiter, clean_str)
-            if (limit is None) or not limit
-            else re.split(options.delimiter, clean_str)[: (limit + 1 if options.raise_on_limit_exceeded else limit)]
-        )
+        _all_parts = re.split(options.delimiter, clean_str)
     else:
-        parts = (
-            clean_str.split(options.delimiter)
-            if (limit is None) or not limit
-            else clean_str.split(options.delimiter)[: (limit + 1 if options.raise_on_limit_exceeded else limit)]
-        )
+        _all_parts = clean_str.split(options.delimiter)
+
+    if (limit is not None) and limit:
+        _take = limit + 1 if options.raise_on_limit_exceeded else limit
+        parts = _all_parts[:_take]
+    else:
+        parts = _all_parts
 
     # Enforce parameter count when strict mode is enabled.
     if options.raise_on_limit_exceeded and (limit is not None) and len(parts) > limit:
@@ -232,9 +230,10 @@ def _parse_query_string_values(value: str, options: DecodeOptions) -> t.Dict[str
                 val if isinstance(val, str) else ",".join(val) if isinstance(val, (list, tuple)) else str(val)
             )
 
-        # If the pair used empty brackets syntax, note that as a list marker.
-        if "[]=" in part:
-            val = [val] if isinstance(val, (list, tuple)) else val
+        # If the pair used empty brackets syntax and list parsing is enabled, force an array container.
+        # Always wrap exactly once to preserve list-of-lists semantics when comma splitting applies.
+        if options.parse_lists and "[]=" in part:
+            val = [val]
 
         existing: bool = key in obj
 
@@ -303,7 +302,10 @@ def _parse_object(
             # Optionally treat `%2E` as a literal dot (when `decode_dot_in_keys` is enabled).
             clean_root: str = root[1:-1] if root.startswith("[") and root.endswith("]") else root
 
-            decoded_root: str = clean_root.replace(r"%2E", ".") if options.decode_dot_in_keys else clean_root
+            if options.decode_dot_in_keys:
+                decoded_root: str = clean_root.replace("%2E", ".").replace("%2e", ".")
+            else:
+                decoded_root = clean_root
 
             # Parse numeric segment to decide between dict key vs. list index.
             index: t.Optional[int]
@@ -322,7 +324,7 @@ def _parse_object(
                 and options.parse_lists
                 and index <= options.list_limit
             ):
-                obj = [Undefined() for _ in range(index + 1)]
+                obj = [UNDEFINED for _ in range(index + 1)]
                 obj[index] = leaf
             else:
                 obj[str(index) if index is not None else decoded_root] = leaf

src/qs_codec/encode.py

@@ -24,7 +24,7 @@
 from .enums.list_format import ListFormat
 from .enums.sentinel import Sentinel
 from .models.encode_options import EncodeOptions
-from .models.undefined import Undefined
+from .models.undefined import UNDEFINED, Undefined
 from .models.weak_wrapper import WeakWrapper
 from .utils.utils import Utils
 
@@ -60,15 +60,15 @@ def encode(value: t.Any, options: EncodeOptions = EncodeOptions()) -> str:
     if isinstance(value, t.Mapping):
         obj = deepcopy(value)
     elif isinstance(value, (list, tuple)):
-        obj = {str(key): value for key, value in enumerate(value)}
+        obj = {str(i): item for i, item in enumerate(value)}
     else:
         obj = {}
 
     # Early exit if there's nothing to emit.
     if not obj:
         return ""
 
-    keys: t.List[t.Any] = []
+    keys: t.List[str] = []
 
     # If an iterable filter is provided for the root, restrict emission to those keys.
     obj_keys: t.Optional[t.List[t.Any]] = None
@@ -218,14 +218,14 @@ def _encode(
 
     # Infer comma round-trip when using the COMMA generator and the flag was not explicitly provided.
     if comma_round_trip is None:
-        comma_round_trip = generate_array_prefix == ListFormat.COMMA.generator
+        comma_round_trip = generate_array_prefix is ListFormat.COMMA.generator
 
     # Choose a formatter if one wasn't provided (based on the selected format).
     if formatter is None:
         formatter = format.formatter
 
-    # Work on a copy to avoid mutating caller state during normalization.
-    obj: t.Any = deepcopy(value)
+    # Work with the original; we never mutate in place (we build new lists/maps when normalizing).
+    obj: t.Any = value
 
     # --- Cycle detection via chained side-channel -----------------------------------------
     obj_wrapper: WeakWrapper = WeakWrapper(value)
@@ -255,7 +255,7 @@ def _encode(
         # Normalize datetimes both for scalars and (in COMMA mode) list elements.
         if isinstance(obj, datetime):
             obj = serialize_date(obj) if callable(serialize_date) else obj.isoformat()
-        elif generate_array_prefix == ListFormat.COMMA.generator and isinstance(obj, (list, tuple)):
+        elif generate_array_prefix is ListFormat.COMMA.generator and isinstance(obj, (list, tuple)):
             if callable(serialize_date):
                 obj = [serialize_date(x) if isinstance(x, datetime) else x for x in obj]
             else:
@@ -289,23 +289,22 @@ def _encode(
         if encode_values_only and callable(encoder):
             obj = Utils.apply(obj, encoder)
         if obj:
-            obj_keys_value = ",".join([str(e) if e is not None else "" for e in obj])
+            obj_keys_value = ",".join(("" if e is None else str(e)) for e in obj)
             obj_keys = [{"value": obj_keys_value if obj_keys_value else None}]
         else:
-            obj_keys = [{"value": Undefined()}]
+            obj_keys = [{"value": UNDEFINED}]
     elif isinstance(filter, (list, tuple)):
         # Iterable filter restricts traversal to a fixed key/index set.
         obj_keys = list(filter)
     else:
         # Default: enumerate keys/indices from mappings or sequences.
-        keys: t.List[t.Any]
         if isinstance(obj, t.Mapping):
             keys = list(obj.keys())
         elif isinstance(obj, (list, tuple)):
-            keys = [index for index in range(len(obj))]
+            keys = list(range(len(obj)))
         else:
             keys = []
-        obj_keys = sorted(keys, key=cmp_to_key(sort)) if sort is not None else list(keys)
+        obj_keys = sorted(keys, key=cmp_to_key(sort)) if sort is not None else keys
 
     # Percent-encode literal dots in key names when requested.
     encoded_prefix: str = prefix.replace(".", "%2E") if encode_dot_in_keys else prefix
@@ -372,7 +371,7 @@ def _encode(
             comma_round_trip=comma_round_trip,
             encoder=(
                 None
-                if generate_array_prefix == ListFormat.COMMA.generator
+                if generate_array_prefix is ListFormat.COMMA.generator
                 and encode_values_only
                 and isinstance(obj, (list, tuple))
                 else encoder

src/qs_codec/models/decode_options.py

@@ -126,7 +126,12 @@ class DecodeOptions:
 
     def __post_init__(self) -> None:
         """Post-initialization."""
-        if self.allow_dots is None:
-            self.allow_dots = self.decode_dot_in_keys is True or False
+        # Default `decode_dot_in_keys` first, then mirror into `allow_dots` when unspecified.
         if self.decode_dot_in_keys is None:
             self.decode_dot_in_keys = False
+        if self.allow_dots is None:
+            self.allow_dots = bool(self.decode_dot_in_keys)
+
+        # Enforce consistency with the docs: `decode_dot_in_keys=True` implies `allow_dots=True`.
+        if self.decode_dot_in_keys and not self.allow_dots:
+            raise ValueError("decode_dot_in_keys=True implies allow_dots=True")

src/qs_codec/models/encode_options.py

@@ -12,7 +12,7 @@
 """
 
 import typing as t
-from dataclasses import asdict, dataclass, field
+from dataclasses import dataclass, field, fields
 from datetime import datetime
 
 from ..enums.charset import Charset
@@ -122,10 +122,12 @@ def __post_init__(self) -> None:
         """
         if not hasattr(self, "_encoder") or self._encoder is None:
             self._encoder = EncodeUtils.encode
-        if self.allow_dots is None:
-            self.allow_dots = self.encode_dot_in_keys is True or False
+        # Default `encode_dot_in_keys` first, then mirror into `allow_dots` when unspecified.
         if self.encode_dot_in_keys is None:
             self.encode_dot_in_keys = False
+        if self.allow_dots is None:
+            self.allow_dots = bool(self.encode_dot_in_keys)
+        # Map deprecated `indices` to `list_format` for backward compatibility.
         if self.indices is not None:
             self.list_format = ListFormat.INDICES if self.indices else ListFormat.REPEAT
 
@@ -139,10 +141,14 @@ def __eq__(self, other: object) -> bool:
         if not isinstance(other, EncodeOptions):
             return False
 
-        self_dict = asdict(self)
-        other_dict = asdict(other)
-
-        self_dict["encoder"] = self._encoder
-        other_dict["encoder"] = other._encoder
-
-        return self_dict == other_dict
+        for f in fields(EncodeOptions):
+            name = f.name
+            if name == "encoder":
+                v1 = self._encoder
+                v2 = other._encoder
+            else:
+                v1 = getattr(self, name)
+                v2 = getattr(other, name)
+            if v1 != v2:
+                return False
+        return True

src/qs_codec/models/undefined.py

@@ -10,6 +10,7 @@
 (e.g., `value is Undefined()`).
 """
 
+import threading
 import typing as t
 
 
@@ -32,6 +33,8 @@ class Undefined:
         ...     pass  # skip emitting the key
     """
 
+    __slots__ = ()
+    _lock: t.ClassVar[threading.Lock] = threading.Lock()
     _instance: t.ClassVar[t.Optional["Undefined"]] = None
 
     def __new__(cls: t.Type["Undefined"]) -> "Undefined":
@@ -40,5 +43,31 @@ def __new__(cls: t.Type["Undefined"]) -> "Undefined":
         Creating `Undefined()` multiple times always returns the same object reference. This ensures identity checks (``is``) are stable.
         """
         if cls._instance is None:
-            cls._instance = super().__new__(cls)
+            with cls._lock:
+                if cls._instance is None:
+                    cls._instance = super().__new__(cls)
         return cls._instance
+
+    def __repr__(self) -> str:  # pragma: no cover - trivial
+        """Return a string representation of the singleton."""
+        return "Undefined()"
+
+    def __copy__(self):  # pragma: no cover - trivial
+        """Ensure copies/pickles preserve the singleton identity."""
+        return self
+
+    def __deepcopy__(self, memo):  # pragma: no cover - trivial
+        """Ensure deep copies preserve the singleton identity."""
+        return self
+
+    def __reduce__(self):  # pragma: no cover - trivial
+        """Recreate via calling the constructor, which returns the singleton."""
+        return Undefined, ()
+
+    def __init_subclass__(cls, **kwargs):  # pragma: no cover - defensive
+        """Prevent subclassing of Undefined."""
+        raise TypeError("Undefined cannot be subclassed")
+
+
+UNDEFINED: t.Final["Undefined"] = Undefined()
+__all__ = ["Undefined", "UNDEFINED"]

src/qs_codec/models/weak_wrapper.py

@@ -30,9 +30,6 @@ def __init__(self, value: t.Any) -> None:
 def _get_proxy(value: t.Any) -> "_Proxy":
     """Return a per-object proxy, cached by id(value)."""
     key = id(value)
-    proxy = _proxy_cache.get(key)
-    if proxy is not None:
-        return proxy
     with _proxy_cache_lock:
         proxy = _proxy_cache.get(key)
         if proxy is None:
@@ -72,14 +69,10 @@ def _leave(oid: int) -> None:
     if isinstance(obj, dict):
         oid = _enter(obj)
         try:
-            # Sort by repr(key) to be stable across runs
-            kv_hashes = tuple(
-                (
-                    _deep_hash(k, _seen, _depth + 1),
-                    _deep_hash(v, _seen, _depth + 1),
-                )
-                for k, v in sorted(obj.items(), key=lambda kv: repr(kv[0]))
-            )
+            # Compute key/value deep hashes once and sort pairs for determinism
+            pairs = [(_deep_hash(k, _seen, _depth + 1), _deep_hash(v, _seen, _depth + 1)) for k, v in obj.items()]
+            pairs.sort()
+            kv_hashes = tuple(pairs)
             return hash(("dict", kv_hashes))
         finally:
             _leave(oid)
@@ -146,8 +139,8 @@ def __eq__(self, other: object) -> bool:
         """Check equality by comparing the proxy identity."""
         if not isinstance(other, WeakWrapper):
             return NotImplemented
-        # Same underlying object => same wrapper identity
-        return self._proxy.value is other._proxy.value
+        # Same underlying object => same cached proxy instance
+        return self._proxy is other._proxy
 
     def __hash__(self) -> int:
         """Return a deep hash of the wrapped value."""

src/qs_codec/utils/decode_utils.py

@@ -31,10 +31,12 @@ class DecodeUtils:
     # "a.b[c]" becomes "a[b][c]" before bracket parsing.
     DOT_TO_BRACKET: t.Pattern[str] = re.compile(r"\.([^.\[]+)")
 
+    # Precompiled pattern for %XX hex bytes (Latin-1 path fast path)
+    HEX2_PATTERN: t.Pattern[str] = re.compile(r"%([0-9A-Fa-f]{2})")
+
     @classmethod
     def unescape(cls, string: str) -> str:
-        """
-        Emulate legacy JavaScript unescape behavior.
+        """Emulate legacy JavaScript unescape behavior.
 
         Replaces both ``%XX`` and ``%uXXXX`` escape sequences with the
         corresponding code points. This function is intentionally permissive
@@ -48,6 +50,9 @@ def unescape(cls, string: str) -> str:
         >>> DecodeUtils.unescape("%7E")
         '~'
         """
+        # Fast path: nothing to unescape
+        if "%" not in string:
+            return string
 
         def replacer(match: t.Match[str]) -> str:
             if (unicode_val := match.group("unicode")) is not None:
@@ -81,18 +86,18 @@ def decode(
         if string is None:
             return None
 
-        # Replace '+' with ' ' before processing.
-        string_without_plus: str = string.replace("+", " ")
+        # Replace '+' with ' ' only if present to avoid allocation.
+        string_without_plus: str = string.replace("+", " ") if "+" in string else string
 
         if charset == Charset.LATIN1:
-            # Only process hex escape sequences for Latin1.
-            return re.sub(
-                r"%[0-9A-Fa-f]{2}",
-                lambda match: cls.unescape(match.group(0)),
-                string_without_plus,
-            )
+            # Only process %XX hex escape sequences for Latin-1 (no %uXXXX expansion here).
+            s = string_without_plus
+            if "%" not in s:
+                return s
+            return cls.HEX2_PATTERN.sub(lambda m: chr(int(m.group(1), 16)), s)
 
-        return unquote(string_without_plus)
+        s = string_without_plus
+        return s if "%" not in s else unquote(s)
 
     @classmethod
     def split_key_into_segments(
@@ -102,8 +107,7 @@ def split_key_into_segments(
         max_depth: int,
         strict_depth: bool,
     ) -> t.List[str]:
-        """
-        Split a composite key into *balanced* bracket segments.
+        """Split a composite key into *balanced* bracket segments.
 
         - If ``allow_dots`` is True, convert dots to bracket groups first (``a.b[c]`` → ``a[b][c]``) while leaving existing brackets intact.
         - The *parent* (non‑bracket) prefix becomes the first segment, e.g. ``"a[b][c]"`` → ``["a", "[b]", "[c]"]``.
@@ -113,7 +117,10 @@ def split_key_into_segments(
 
         This runs in O(n) time over the key string.
         """
-        key: str = cls.DOT_TO_BRACKET.sub(r"[\1]", original_key) if allow_dots else original_key
+        if allow_dots and "." in original_key:
+            key: str = cls.DOT_TO_BRACKET.sub(r"[\1]", original_key)
+        else:
+            key = original_key
 
         if max_depth <= 0:
             return [key]