1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
|
"""Minor internal utility functions that don't really belong anywhere else"""
from contextlib import contextmanager
from inspect import signature
from logging import getLogger
from typing import Any, Callable, Dict, Iterable, Iterator, List, Optional, Tuple
from urllib3 import filepost
FORM_BOUNDARY = '--requests-cache-form-boundary--'
KwargDict = Dict[str, Any]
logger = getLogger('requests_cache')
def chunkify(iterable: Optional[Iterable], max_size: int) -> Iterator[List]:
"""Split an iterable into chunks of a max size"""
iterable = list(iterable or [])
for index in range(0, len(iterable), max_size):
yield iterable[index : index + max_size]
def coalesce(*values: Any, default=None) -> Any:
"""Get the first non-``None`` value in a list of values"""
return next((v for v in values if v is not None), default)
def decode(value, encoding='utf-8') -> str:
"""Decode a value from bytes, if hasn't already been.
Note: ``PreparedRequest.body`` is always encoded in utf-8.
"""
if not value:
return ''
return value.decode(encoding) if isinstance(value, bytes) else value
def encode(value, encoding='utf-8') -> bytes:
"""Encode a value to bytes, if it hasn't already been"""
if not value:
return b''
return value if isinstance(value, bytes) else str(value).encode(encoding)
def get_placeholder_class(original_exception: Optional[Exception] = None):
"""Create a placeholder type for a class that does not have dependencies installed.
This allows delaying ImportErrors until init time, rather than at import time.
"""
def _log_error():
msg = 'Dependencies are not installed for this feature'
logger.error(msg)
raise original_exception or ImportError(msg)
class Placeholder:
def __init__(self, *args, **kwargs):
_log_error()
def dumps(self, *args, **kwargs):
_log_error()
def loads(self, *args, **kwargs):
_log_error()
return Placeholder
def get_valid_kwargs(
func: Callable, kwargs: Dict, extras: Optional[Iterable[str]] = None
) -> KwargDict:
"""Get the subset of non-None ``kwargs`` that are valid arguments for ``func``"""
kwargs, _ = split_kwargs(func, kwargs, extras)
return {k: v for k, v in kwargs.items() if v is not None}
@contextmanager
def patch_form_boundary():
"""If the ``files`` param is present, patch the form boundary used to separate multipart
uploads. ``requests`` does not provide a way to pass a custom boundary to urllib3, so this just
monkey-patches it instead.
"""
original_boundary = filepost.choose_boundary
filepost.choose_boundary = lambda: FORM_BOUNDARY
yield
filepost.choose_boundary = original_boundary
def split_kwargs(
func: Callable, kwargs: Dict, extras: Optional[Iterable[str]] = None
) -> Tuple[KwargDict, KwargDict]:
"""Split ``kwargs`` into two dicts: those that are valid arguments for ``func``, and those that
are not
"""
params = list(signature(func).parameters)
params.extend(extras or [])
valid_kwargs = {k: v for k, v in kwargs.items() if k in params}
invalid_kwargs = {k: v for k, v in kwargs.items() if k not in params}
return valid_kwargs, invalid_kwargs
def try_int(value: Any) -> Optional[int]:
"""Convert a value to an int, if possible, otherwise ``None``"""
try:
return int(value)
except (TypeError, ValueError):
return None
def is_json_content_type(content_type: Optional[str]) -> bool:
"""Returns whether the given content-type represents json"""
# empiric solution to catch stuff like `application/json;charset=UTF-8` or `application/vnd.api+json`
return bool(content_type and content_type.startswith('application/') and 'json' in content_type)
|
