diff options
| author | Jordan Cook <jordan.cook@pioneer.com> | 2021-10-26 16:37:39 -0500 |
|---|---|---|
| committer | Jordan Cook <jordan.cook@pioneer.com> | 2021-10-27 10:35:39 -0500 |
| commit | 90d2b87c14c990c732b52e6abea907e4ef86166f (patch) | |
| tree | 9ff01161129289fab06383b72300504ae2724f7b /requests_cache/_utils.py | |
| parent | a70eea1beadcb2d555a7dac77738d0a21900703a (diff) | |
| download | requests-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.py | 48 |
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} |
