feat: implement QueryParams class with methods for handling URL query parameters

This commit is contained in:
Бакыт Ниязалиев 2025-06-13 04:09:14 +07:00
parent 44897aa3fc
commit 9fc20e646c
3 changed files with 330 additions and 202 deletions

View File

@ -0,0 +1,152 @@
import typing
PrimitiveData = typing.Optional[typing.Union[str, int, float, bool]]
QueryParamTypes = typing.Union[
"QueryParams",
typing.Mapping[str, typing.Union[PrimitiveData, typing.Sequence[PrimitiveData]]],
typing.List[typing.Tuple[str, PrimitiveData]],
typing.Tuple[typing.Tuple[str, PrimitiveData], ...],
str,
bytes,
]
class QueryParams(typing.Mapping[str, str]):
def __init__(self, *args: QueryParamTypes | None, **kwargs: typing.Any) -> None:...
def keys(self) -> typing.KeysView[str]:
"""
Return all the keys in the query params.
Usage:
```
q = httpx.QueryParams("a=123&a=456&b=789")
assert list(q.keys()) == ["a", "b"]
```
"""
def values(self) -> typing.ValuesView[str]:
"""
Return all the values in the query params. If a key occurs more than once
only the first item for that key is returned.
Usage:
```
q = httpx.QueryParams("a=123&a=456&b=789")
assert list(q.values()) == ["123", "789"]
```
"""
def items(self) -> typing.ItemsView[str, str]:
"""
Return all items in the query params. If a key occurs more than once
only the first item for that key is returned.
Usage:
q = httpx.QueryParams("a=123&a=456&b=789")
assert list(q.items()) == [("a", "123"), ("b", "789")]
"""
def multi_items(self) -> list[tuple[str, str]]:
"""
Return all items in the query params. Allow duplicate keys to occur.
Usage:
```
q = httpx.QueryParams("a=123&a=456&b=789")
assert list(q.multi_items()) == [("a", "123"), ("a", "456"), ("b", "789")]
```
"""
def get(self, key: typing.Any, default: typing.Any = None) -> typing.Any:
"""
Get a value from the query param for a given key. If the key occurs
more than once, then only the first value is returned.
Usage:
```
q = httpx.QueryParams("a=123&a=456&b=789")
assert q.get("a") == "123"
```
"""
def get_list(self, key: str) -> list[str]:
"""
Get all values from the query param for a given key.
Usage:
```
q = httpx.QueryParams("a=123&a=456&b=789")
assert q.get_list("a") == ["123", "456"]
```
"""
def set(self, key: str, value: typing.Any = None) -> QueryParams:
"""
Return a new QueryParams instance, setting the value of a key.
Usage:
```
q = httpx.QueryParams("a=123")
q = q.set("a", "456")
assert q == httpx.QueryParams("a=456")
```
"""
def add(self, key: str, value: typing.Any = None) -> QueryParams:
"""
Return a new QueryParams instance, setting or appending the value of a key.
Usage:
```
q = httpx.QueryParams("a=123")
q = q.add("a", "456")
assert q == httpx.QueryParams("a=123&a=456")
```
"""
def remove(self, key: str) -> QueryParams:
"""
Return a new QueryParams instance, removing the value of a key.
Usage:
```
q = httpx.QueryParams("a=123")
q = q.remove("a")
assert q == httpx.QueryParams("")
```
"""
def merge(self, params: QueryParamTypes | None = None) -> QueryParams:
"""
Return a new QueryParams instance, updated with.
Usage:
```
q = httpx.QueryParams("a=123")
q = q.merge({"b": "456"})
assert q == httpx.QueryParams("a=123&b=456")
q = httpx.QueryParams("a=123")
q = q.merge({"a": "456", "b": "789"})
assert q == httpx.QueryParams("a=456&b=789")
```
"""
def __getitem__(self, key: typing.Any) -> str:...
def __contains__(self, key: typing.Any) -> bool:...
def __iter__(self) -> typing.Iterator[typing.Any]:...
def __len__(self) -> int:...
def __bool__(self) -> bool:...
def __hash__(self) -> int:...
def __eq__(self, other: typing.Any) -> bool:...
def __str__(self) -> str:...
def __repr__(self) -> str:...
def update(self, params: QueryParamTypes | None = None) -> None:...
def __setitem__(self, key: str, value: str) -> None:...

View File

@ -1,24 +0,0 @@
class ParseResult:
scheme: str
userinfo: str
host: str
port: int | None
path: str
query: str | None
fragment: str | None
@property
def authority(self) -> str: ...
@property
def netloc(self) -> str: ...
def __str__(self) -> str: ...
def __new__(
cls,
scheme: str,
userinfo: str,
host: str,
port: int | None,
path: str,
query: str | None,
fragment: str | None,
) -> ParseResult: ...

View File

@ -1,13 +1,13 @@
from __future__ import annotations
import typing
from urllib.parse import parse_qs, unquote, urlencode
from urllib.parse import unquote
import idna
from ._httpx import QueryParams
from ._types import QueryParamTypes
from ._urlparse import urlparse
from ._utils import primitive_value_to_str
__all__ = ["URL", "QueryParams"]
@ -417,225 +417,225 @@ class URL:
)
class QueryParams(typing.Mapping[str, str]):
"""
URL query parameters, as a multi-dict.
"""
# class QueryParams(typing.Mapping[str, str]):
# """
# URL query parameters, as a multi-dict.
# """
def __init__(self, *args: QueryParamTypes | None, **kwargs: typing.Any) -> None:
assert len(args) < 2, "Too many arguments."
assert not (args and kwargs), "Cannot mix named and unnamed arguments."
# def __init__(self, *args: QueryParamTypes | None, **kwargs: typing.Any) -> None:
# assert len(args) < 2, "Too many arguments."
# assert not (args and kwargs), "Cannot mix named and unnamed arguments."
value = args[0] if args else kwargs
# value = args[0] if args else kwargs
if value is None or isinstance(value, (str, bytes)):
value = value.decode("ascii") if isinstance(value, bytes) else value
self._dict = parse_qs(value, keep_blank_values=True)
elif isinstance(value, QueryParams):
self._dict = {k: list(v) for k, v in value._dict.items()}
else:
dict_value: dict[typing.Any, list[typing.Any]] = {}
if isinstance(value, (list, tuple)):
# Convert list inputs like:
# [("a", "123"), ("a", "456"), ("b", "789")]
# To a dict representation, like:
# {"a": ["123", "456"], "b": ["789"]}
for item in value:
dict_value.setdefault(item[0], []).append(item[1])
else:
# Convert dict inputs like:
# {"a": "123", "b": ["456", "789"]}
# To dict inputs where values are always lists, like:
# {"a": ["123"], "b": ["456", "789"]}
dict_value = {
k: list(v) if isinstance(v, (list, tuple)) else [v]
for k, v in value.items()
}
# if value is None or isinstance(value, (str, bytes)):
# value = value.decode("ascii") if isinstance(value, bytes) else value
# self._dict = parse_qs(value, keep_blank_values=True)
# elif isinstance(value, QueryParams):
# self._dict = {k: list(v) for k, v in value._dict.items()}
# else:
# dict_value: dict[typing.Any, list[typing.Any]] = {}
# if isinstance(value, (list, tuple)):
# # Convert list inputs like:
# # [("a", "123"), ("a", "456"), ("b", "789")]
# # To a dict representation, like:
# # {"a": ["123", "456"], "b": ["789"]}
# for item in value:
# dict_value.setdefault(item[0], []).append(item[1])
# else:
# # Convert dict inputs like:
# # {"a": "123", "b": ["456", "789"]}
# # To dict inputs where values are always lists, like:
# # {"a": ["123"], "b": ["456", "789"]}
# dict_value = {
# k: list(v) if isinstance(v, (list, tuple)) else [v]
# for k, v in value.items()
# }
# Ensure that keys and values are neatly coerced to strings.
# We coerce values `True` and `False` to JSON-like "true" and "false"
# representations, and coerce `None` values to the empty string.
self._dict = {
str(k): [primitive_value_to_str(item) for item in v]
for k, v in dict_value.items()
}
# # Ensure that keys and values are neatly coerced to strings.
# # We coerce values `True` and `False` to JSON-like "true" and "false"
# # representations, and coerce `None` values to the empty string.
# self._dict = {
# str(k): [primitive_value_to_str(item) for item in v]
# for k, v in dict_value.items()
# }
def keys(self) -> typing.KeysView[str]:
"""
Return all the keys in the query params.
# def keys(self) -> typing.KeysView[str]:
# """
# Return all the keys in the query params.
Usage:
# Usage:
q = httpx.QueryParams("a=123&a=456&b=789")
assert list(q.keys()) == ["a", "b"]
"""
return self._dict.keys()
# q = httpx.QueryParams("a=123&a=456&b=789")
# assert list(q.keys()) == ["a", "b"]
# """
# return self._dict.keys()
def values(self) -> typing.ValuesView[str]:
"""
Return all the values in the query params. If a key occurs more than once
only the first item for that key is returned.
# def values(self) -> typing.ValuesView[str]:
# """
# Return all the values in the query params. If a key occurs more than once
# only the first item for that key is returned.
Usage:
# Usage:
q = httpx.QueryParams("a=123&a=456&b=789")
assert list(q.values()) == ["123", "789"]
"""
return {k: v[0] for k, v in self._dict.items()}.values()
# q = httpx.QueryParams("a=123&a=456&b=789")
# assert list(q.values()) == ["123", "789"]
# """
# return {k: v[0] for k, v in self._dict.items()}.values()
def items(self) -> typing.ItemsView[str, str]:
"""
Return all items in the query params. If a key occurs more than once
only the first item for that key is returned.
# def items(self) -> typing.ItemsView[str, str]:
# """
# Return all items in the query params. If a key occurs more than once
# only the first item for that key is returned.
Usage:
# Usage:
q = httpx.QueryParams("a=123&a=456&b=789")
assert list(q.items()) == [("a", "123"), ("b", "789")]
"""
return {k: v[0] for k, v in self._dict.items()}.items()
# q = httpx.QueryParams("a=123&a=456&b=789")
# assert list(q.items()) == [("a", "123"), ("b", "789")]
# """
# return {k: v[0] for k, v in self._dict.items()}.items()
def multi_items(self) -> list[tuple[str, str]]:
"""
Return all items in the query params. Allow duplicate keys to occur.
# def multi_items(self) -> list[tuple[str, str]]:
# """
# Return all items in the query params. Allow duplicate keys to occur.
Usage:
# Usage:
q = httpx.QueryParams("a=123&a=456&b=789")
assert list(q.multi_items()) == [("a", "123"), ("a", "456"), ("b", "789")]
"""
multi_items: list[tuple[str, str]] = []
for k, v in self._dict.items():
multi_items.extend([(k, i) for i in v])
return multi_items
# q = httpx.QueryParams("a=123&a=456&b=789")
# assert list(q.multi_items()) == [("a", "123"), ("a", "456"), ("b", "789")]
# """
# multi_items: list[tuple[str, str]] = []
# for k, v in self._dict.items():
# multi_items.extend([(k, i) for i in v])
# return multi_items
def get(self, key: typing.Any, default: typing.Any = None) -> typing.Any:
"""
Get a value from the query param for a given key. If the key occurs
more than once, then only the first value is returned.
# def get(self, key: typing.Any, default: typing.Any = None) -> typing.Any:
# """
# Get a value from the query param for a given key. If the key occurs
# more than once, then only the first value is returned.
Usage:
# Usage:
q = httpx.QueryParams("a=123&a=456&b=789")
assert q.get("a") == "123"
"""
if key in self._dict:
return self._dict[str(key)][0]
return default
# q = httpx.QueryParams("a=123&a=456&b=789")
# assert q.get("a") == "123"
# """
# if key in self._dict:
# return self._dict[str(key)][0]
# return default
def get_list(self, key: str) -> list[str]:
"""
Get all values from the query param for a given key.
# def get_list(self, key: str) -> list[str]:
# """
# Get all values from the query param for a given key.
Usage:
# Usage:
q = httpx.QueryParams("a=123&a=456&b=789")
assert q.get_list("a") == ["123", "456"]
"""
return list(self._dict.get(str(key), []))
# q = httpx.QueryParams("a=123&a=456&b=789")
# assert q.get_list("a") == ["123", "456"]
# """
# return list(self._dict.get(str(key), []))
def set(self, key: str, value: typing.Any = None) -> QueryParams:
"""
Return a new QueryParams instance, setting the value of a key.
# def set(self, key: str, value: typing.Any = None) -> QueryParams:
# """
# Return a new QueryParams instance, setting the value of a key.
Usage:
# Usage:
q = httpx.QueryParams("a=123")
q = q.set("a", "456")
assert q == httpx.QueryParams("a=456")
"""
q = QueryParams()
q._dict = dict(self._dict)
q._dict[str(key)] = [primitive_value_to_str(value)]
return q
# q = httpx.QueryParams("a=123")
# q = q.set("a", "456")
# assert q == httpx.QueryParams("a=456")
# """
# q = QueryParams()
# q._dict = dict(self._dict)
# q._dict[str(key)] = [primitive_value_to_str(value)]
# return q
def add(self, key: str, value: typing.Any = None) -> QueryParams:
"""
Return a new QueryParams instance, setting or appending the value of a key.
# def add(self, key: str, value: typing.Any = None) -> QueryParams:
# """
# Return a new QueryParams instance, setting or appending the value of a key.
Usage:
# Usage:
q = httpx.QueryParams("a=123")
q = q.add("a", "456")
assert q == httpx.QueryParams("a=123&a=456")
"""
q = QueryParams()
q._dict = dict(self._dict)
q._dict[str(key)] = q.get_list(key) + [primitive_value_to_str(value)]
return q
# q = httpx.QueryParams("a=123")
# q = q.add("a", "456")
# assert q == httpx.QueryParams("a=123&a=456")
# """
# q = QueryParams()
# q._dict = dict(self._dict)
# q._dict[str(key)] = q.get_list(key) + [primitive_value_to_str(value)]
# return q
def remove(self, key: str) -> QueryParams:
"""
Return a new QueryParams instance, removing the value of a key.
# def remove(self, key: str) -> QueryParams:
# """
# Return a new QueryParams instance, removing the value of a key.
Usage:
# Usage:
q = httpx.QueryParams("a=123")
q = q.remove("a")
assert q == httpx.QueryParams("")
"""
q = QueryParams()
q._dict = dict(self._dict)
q._dict.pop(str(key), None)
return q
# q = httpx.QueryParams("a=123")
# q = q.remove("a")
# assert q == httpx.QueryParams("")
# """
# q = QueryParams()
# q._dict = dict(self._dict)
# q._dict.pop(str(key), None)
# return q
def merge(self, params: QueryParamTypes | None = None) -> QueryParams:
"""
Return a new QueryParams instance, updated with.
# def merge(self, params: QueryParamTypes | None = None) -> QueryParams:
# """
# Return a new QueryParams instance, updated with.
Usage:
# Usage:
q = httpx.QueryParams("a=123")
q = q.merge({"b": "456"})
assert q == httpx.QueryParams("a=123&b=456")
# q = httpx.QueryParams("a=123")
# q = q.merge({"b": "456"})
# assert q == httpx.QueryParams("a=123&b=456")
q = httpx.QueryParams("a=123")
q = q.merge({"a": "456", "b": "789"})
assert q == httpx.QueryParams("a=456&b=789")
"""
q = QueryParams(params)
q._dict = {**self._dict, **q._dict}
return q
# q = httpx.QueryParams("a=123")
# q = q.merge({"a": "456", "b": "789"})
# assert q == httpx.QueryParams("a=456&b=789")
# """
# q = QueryParams(params)
# q._dict = {**self._dict, **q._dict}
# return q
def __getitem__(self, key: typing.Any) -> str:
return self._dict[key][0]
# def __getitem__(self, key: typing.Any) -> str:
# return self._dict[key][0]
def __contains__(self, key: typing.Any) -> bool:
return key in self._dict
# def __contains__(self, key: typing.Any) -> bool:
# return key in self._dict
def __iter__(self) -> typing.Iterator[typing.Any]:
return iter(self.keys())
# def __iter__(self) -> typing.Iterator[typing.Any]:
# return iter(self.keys())
def __len__(self) -> int:
return len(self._dict)
# def __len__(self) -> int:
# return len(self._dict)
def __bool__(self) -> bool:
return bool(self._dict)
# def __bool__(self) -> bool:
# return bool(self._dict)
def __hash__(self) -> int:
return hash(str(self))
# def __hash__(self) -> int:
# return hash(str(self))
def __eq__(self, other: typing.Any) -> bool:
if not isinstance(other, self.__class__):
return False
return sorted(self.multi_items()) == sorted(other.multi_items())
# def __eq__(self, other: typing.Any) -> bool:
# if not isinstance(other, self.__class__):
# return False
# return sorted(self.multi_items()) == sorted(other.multi_items())
def __str__(self) -> str:
return urlencode(self.multi_items())
# def __str__(self) -> str:
# return urlencode(self.multi_items())
def __repr__(self) -> str:
class_name = self.__class__.__name__
query_string = str(self)
return f"{class_name}({query_string!r})"
# def __repr__(self) -> str:
# class_name = self.__class__.__name__
# query_string = str(self)
# return f"{class_name}({query_string!r})"
def update(self, params: QueryParamTypes | None = None) -> None:
raise RuntimeError(
"QueryParams are immutable since 0.18.0. "
"Use `q = q.merge(...)` to create an updated copy."
)
# def update(self, params: QueryParamTypes | None = None) -> None:
# raise RuntimeError(
# "QueryParams are immutable since 0.18.0. "
# "Use `q = q.merge(...)` to create an updated copy."
# )
def __setitem__(self, key: str, value: str) -> None:
raise RuntimeError(
"QueryParams are immutable since 0.18.0. "
"Use `q = q.set(key, value)` to create an updated copy."
)
# def __setitem__(self, key: str, value: str) -> None:
# raise RuntimeError(
# "QueryParams are immutable since 0.18.0. "
# "Use `q = q.set(key, value)` to create an updated copy."
# )