Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 42 additions & 6 deletions benedict/serializers/query_string.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,17 +26,53 @@ def __init__(self) -> None:
],
)

@staticmethod
def _parse_bracket_notation(
data: dict[str, list[str]],
) -> dict[str, Any]:
"""Convert parse_qs output into nested dicts / lists for bracket keys.

- ``a[]=1&a[]=2`` → ``{"a": ["1", "2"]}``
- ``user[name]=joe`` → ``{"user": {"name": "joe"}}``
- Regular keys (no brackets) always return a scalar string value;
the *flat* parameter on :meth:`decode` does not affect them.
"""
_array_re = re.compile(r"^([^\[]+)\[\]$")
_nested_re = re.compile(r"^([^\[]+)\[([^\]]+)\]$")
result: dict[str, Any] = {}
for key, values in data.items():
m_array = _array_re.match(key)
m_nested = _nested_re.match(key)
if m_array:
base = m_array.group(1)
result[base] = values
elif m_nested:
parent, child = m_nested.group(1), m_nested.group(2)
if parent not in result or not isinstance(result[parent], dict):
result[parent] = {}
result[parent][child] = values[0] if len(values) == 1 else values
else:
result[key] = values[0]
return result

@override
def decode( # type: ignore[override]
self, s: str, flat: bool = True
) -> dict[str, str] | dict[str, list[str]]:
qs_re = r"(?:([\w\-\%\+\.\|]+\=[\w\-\%\+\.\|]*)+(?:[\&]{1})?)+"
qs_pattern = re.compile(qs_re)
if qs_pattern.match(s):
# A query string is a sequence of "key=value" pairs joined by "&".
# Each key must be non-empty and free of whitespace, "=" and "&";
# each value must be free of whitespace and "&" (spaces are encoded
# as "+" or "%20"). This accepts real-world keys such as array-style
# "a[]" / "user[name]" while still rejecting other formats (TOML, YAML,
# JSON, XML), plain text and URLs.
pair_re = re.compile(r"^[^\s=&]+=[^\s&]*$")
pairs = s.split("&")
# Allow exactly one trailing empty segment (trailing "&" is valid)
if pairs and not pairs[-1]:
pairs = pairs[:-1]
Comment thread
fabiocaccamo marked this conversation as resolved.
if all(pair_re.match(pair) for pair in pairs):
data = parse_qs(s)
if flat:
return {key: value[0] for key, value in data.items()}
return data
return self._parse_bracket_notation(data)
raise ValueError(f"Invalid query string: {s}")

@override
Expand Down
45 changes: 45 additions & 0 deletions tests/dicts/io/test_io_dict_query_string.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,51 @@ def test_from_query_string_with_valid_data(self) -> None:
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, r)

def test_from_query_string_with_array_style_keys(self) -> None:
# array-style keys (PHP / HTML form syntax) are valid query strings
s = "a[]=1&a[]=2"
r = {"a": ["1", "2"]}
# static method
d = IODict.from_query_string(s)
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, r)
# constructor
d = IODict(s, format="query_string")
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, r)

def test_from_query_string_with_bracketed_keys(self) -> None:
s = "user[name]=joe&user[age]=42"
r = {"user": {"name": "joe", "age": "42"}}
# static method
d = IODict.from_query_string(s)
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, r)

def test_from_query_string_with_trailing_ampersand(self) -> None:
s = "a=1&"
r = {"a": "1"}
# static method
d = IODict.from_query_string(s)
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, r)
# constructor
d = IODict(s, format="query_string")
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, r)

def test_from_query_string_with_valid_data_flat_false(self) -> None:
s = "a=1&b=2"
r = {"a": "1", "b": "2"}
# static method
d = IODict.from_query_string(s, flat=False)
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, r)
# constructor
d = IODict(s, format="query_string", flat=False)
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, r)

def test_from_query_string_with_invalid_data(self) -> None:
s = "Lorem ipsum est in ea occaecat nisi officia."
# static method
Expand Down
Loading