rack_nested_query_parser.py

"""This module implements a parser for query strings encoded with Ruby RestClient:

https://www.rubydoc.info/gems/rest-client/RestClient%2FUtils.encode_query_string
"""
import re
from urllib.parse import unquote


class ParameterTypeError(Exception):
    """Exception for mixed parameter types."""


def parse_legacy_query(query_string: str) -> dict:
    """Rack query parser.

    Reimplementation of
    https://github.com/rack/rack/blob/cd5d902ac801f0bebeeb5d8dc534f717c9822b9b/lib/rack/query_parser.rb#L64

    >>> parse_legacy_query("foo=123&bar=456")
    {'foo': '123', 'bar': '456'}

    >>> parse_legacy_query('foo[]=1&foo[]=2&foo[]=3')
    {'foo': ['1', '2', '3']}

    >>> parse_legacy_query('outer[foo]=123&outer[bar]=456')
    {'outer': {'foo': '123', 'bar': '456'}}

    >>> parse_legacy_query('coords[][x]=1&coords[][y]=0&coords[][x]=2&coords[][x]=3')
    {'coords': [{'x': '1', 'y': '0'}, {'x': '2'}, {'x': '3'}]}

    >>> parse_legacy_query('string=&empty')
    {'string': '', 'empty': None}

    """
    params: dict = {}

    if not query_string:
        return {}
    for part in query_string.split("&"):
        try:
            key, value = unquote(part).split("=", 1)
        except ValueError as exc:
            if "not enough values to unpack" in str(exc):
                key = unquote(part)
                value = None  # type: ignore
            else:
                raise exc
        normalize_params(params, key, value)
    return params


def normalize_params(params, name, value):
    match = re.match(r"^[\[\]]*([^\[\]]+)\]*", name)
    key = match.groups()[0] or ""
    after = name[match.end():] or ""

    if not key:
        raise NotImplementedError(f"Not supported: {key}")

    if not after:
        params[key] = value
    elif after == "[":
        params[name] = value
    elif after == "[]":
        params.setdefault(key, [])
        if not isinstance(params[key], list):
            raise ParameterTypeError(f"Expected list (got {type(params[key])}) for param {key}")
        params[key].append(value)
    elif match := (re.match(r"^\[\]\[([^\[\]]+)\]$", after) or re.match(r"^\[\](.+)$", after)):
        child_key = match.groups()[0]
        params.setdefault(key, [])
        if not isinstance(params[key], list):
            raise ParameterTypeError(f"Expected list (got {type(params[key])}) for param {key}")
        if params[key] and isinstance(params[key][-1], dict) and child_key not in params[key][-1]:
            normalize_params(params[key][-1], child_key, value)
        else:
            sub_params ={}
            normalize_params(sub_params, child_key, value)
            params[key].append(sub_params)
    else:
        params.setdefault(key, {})
        if not isinstance(params[key], dict):
            raise ParameterTypeError(f"Expected dict (got {type(params[key])}) for param {key}")
        normalize_params(params[key], after, value)