|
3 | 3 | import json |
4 | 4 | import re |
5 | 5 | from typing import Optional |
6 | | -from urllib.parse import parse_qs, urlencode |
| 6 | +from urllib.parse import parse_qs |
7 | 7 |
|
8 | 8 | from cql2 import Expr |
9 | 9 |
|
10 | | -from .requests import dict_to_bytes |
11 | | - |
12 | 10 |
|
13 | 11 | def append_qs_filter(qs: str, filter: Expr, filter_lang: Optional[str] = None) -> bytes: |
14 | 12 | """Insert a filter expression into a query string. If a filter already exists, combine them.""" |
15 | 13 | qs_dict = {k: v[0] for k, v in parse_qs(qs).items()} |
16 | 14 | new_qs_dict = append_body_filter( |
17 | 15 | qs_dict, filter, filter_lang or qs_dict.get("filter-lang", "cql2-text") |
18 | 16 | ) |
19 | | - return dict_to_bytes( |
20 | | - urlencode( |
21 | | - { |
22 | | - k: json.dumps(v) if isinstance(v, (list, dict)) else v |
23 | | - for k, v in new_qs_dict.items() |
24 | | - } |
25 | | - ) |
26 | | - ) |
| 17 | + return dict_to_query_string(new_qs_dict).encode("utf-8") |
27 | 18 |
|
28 | 19 |
|
29 | | -def append_body_filter(body: dict, filter: Expr, filter_lang: Optional[str]) -> dict: |
| 20 | +def append_body_filter( |
| 21 | + body: dict, filter: Expr, filter_lang: Optional[str] = None |
| 22 | +) -> dict: |
30 | 23 | """Insert a filter expression into a request body. If a filter already exists, combine them.""" |
31 | 24 | cur_filter = body.get("filter") |
32 | 25 | filter_lang = filter_lang or body.get("filter-lang", "cql2-json") |
@@ -54,3 +47,16 @@ def is_item_endpoint(path: str) -> bool: |
54 | 47 | def is_search_endpoint(path: str) -> bool: |
55 | 48 | """Check if the path is a search endpoint.""" |
56 | 49 | return path == "/search" |
| 50 | + |
| 51 | + |
| 52 | +def dict_to_query_string(params: dict) -> str: |
| 53 | + """ |
| 54 | + Convert a dictionary to a query string. Dict values are converted to JSON strings, |
| 55 | + unlike the default behavior of urllib.parse.urlencode. |
| 56 | + """ |
| 57 | + parts = [] |
| 58 | + for key, val in params.items(): |
| 59 | + if isinstance(val, (dict, list)): |
| 60 | + val = json.dumps(val, separators=(",", ":")) |
| 61 | + parts.append(f"{key}={val}") |
| 62 | + return "&".join(parts) |
0 commit comments