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
|
"""Jira utils used internally."""
from __future__ import annotations
import threading
import warnings
from typing import Any, cast
from requests import Response
from requests.structures import CaseInsensitiveDict as _CaseInsensitiveDict
from jira.resilientsession import raise_on_error
class CaseInsensitiveDict(_CaseInsensitiveDict):
"""A case-insensitive ``dict``-like object.
DEPRECATED: use requests.structures.CaseInsensitiveDict directly.
Implements all methods and operations of
``collections.MutableMapping`` as well as dict's ``copy``. Also
provides ``lower_items``.
All keys are expected to be strings. The structure remembers the
case of the last key to be set, and ``iter(instance)``,
``keys()``, ``items()``, ``iterkeys()``
will contain case-sensitive keys. However, querying and contains
testing is case insensitive::
cid = CaseInsensitiveDict()
cid['Accept'] = 'application/json'
cid['accept'] == 'application/json' # True
list(cid) == ['Accept'] # True
For example, ``headers['content-encoding']`` will return the
value of a ``'Content-Encoding'`` response header, regardless
of how the header name was originally stored.
If the constructor, ``.update``, or equality comparison
operations are given keys that have equal ``.lower()`` s, the
behavior is undefined.
"""
def __init__(self, *args, **kwargs) -> None:
warnings.warn(
"Use requests.structures.CaseInsensitiveDict directly", DeprecationWarning
)
super().__init__(*args, **kwargs)
def threaded_requests(requests):
for fn, url, request_args in requests:
th = threading.Thread(target=fn, args=(url,), kwargs=request_args, name=url)
th.start()
for th in threading.enumerate():
if th.name.startswith("http"):
th.join()
def json_loads(resp: Response | None) -> Any:
"""Attempts to load json the result of a response.
Args:
resp (Optional[Response]): The Response object
Raises:
JIRAError: via :py:func:`jira.resilientsession.raise_on_error`
Returns:
Union[List[Dict[str, Any]], Dict[str, Any]]: the json
"""
raise_on_error(resp) # if 'resp' is None, will raise an error here
resp = cast(Response, resp) # tell mypy only Response-like are here
try:
return resp.json()
except ValueError:
# json.loads() fails with empty bodies
if not resp.text:
return {}
raise
def remove_empty_attributes(data: dict[str, Any]) -> dict[str, Any]:
"""A convenience function to remove key/value pairs with `None` for a value.
Args:
data: A dictionary.
Returns:
Dict[str, Any]: A dictionary with no `None` key/value pairs.
"""
return {key: val for key, val in data.items() if val is not None}
|