summaryrefslogtreecommitdiff
path: root/requests_cache/_utils.py
diff options
context:
space:
mode:
authorJordan Cook <jordan.cook@pioneer.com>2021-10-26 16:37:39 -0500
committerJordan Cook <jordan.cook@pioneer.com>2021-10-27 10:35:39 -0500
commit90d2b87c14c990c732b52e6abea907e4ef86166f (patch)
tree9ff01161129289fab06383b72300504ae2724f7b /requests_cache/_utils.py
parenta70eea1beadcb2d555a7dac77738d0a21900703a (diff)
downloadrequests-cache-90d2b87c14c990c732b52e6abea907e4ef86166f.tar.gz
Move misc minor utils to a separate module
Diffstat (limited to 'requests_cache/_utils.py')
-rw-r--r--requests_cache/_utils.py48
1 files changed, 48 insertions, 0 deletions
diff --git a/requests_cache/_utils.py b/requests_cache/_utils.py
new file mode 100644
index 0000000..b3dfaee
--- /dev/null
+++ b/requests_cache/_utils.py
@@ -0,0 +1,48 @@
+"""Miscellaneous minor utility functions that don't really belong anywhere else"""
+from inspect import signature
+from logging import getLogger
+from typing import Any, Callable, Dict, Iterable, Iterator, List
+
+logger = getLogger('requests_cache')
+
+
+def chunkify(iterable: Iterable, max_size: int) -> Iterator[List]:
+ """Split an iterable into chunks of a max size"""
+ iterable = list(iterable)
+ 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 get_placeholder_class(original_exception: 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.
+ """
+ msg = 'Dependencies are not installed for this feature'
+
+ def _log_error():
+ logger.error(msg)
+ raise original_exception or ImportError(msg)
+
+ class Placeholder:
+ def __init__(self, *args, **kwargs):
+ _log_error()
+
+ def __getattr__(self, *args, **kwargs):
+ _log_error()
+
+ def dumps(self, *args, **kwargs):
+ _log_error()
+
+ return Placeholder
+
+
+def get_valid_kwargs(func: Callable, kwargs: Dict, extras: Iterable[str] = None) -> Dict:
+ """Get the subset of non-None ``kwargs`` that are valid params for ``func``"""
+ params = list(signature(func).parameters)
+ params.extend(extras or [])
+ return {k: v for k, v in kwargs.items() if k in params and v is not None}