This document describes the current stable version of Celery (5.2). For development docs, go here.

Source code for celery.utils.functional

"""Functional-style utilties."""
import inspect
import sys
from collections import UserList
from functools import partial
from itertools import islice, tee, zip_longest

from kombu.utils.functional import (LRUCache, dictfilter, is_list, lazy,
                                    maybe_evaluate, maybe_list, memoize)
from vine import promise

__all__ = (
    'LRUCache', 'is_list', 'maybe_list', 'memoize', 'mlazy', 'noop',
    'first', 'firstmethod', 'chunks', 'padlist', 'mattrgetter', 'uniq',
    'regen', 'dictfilter', 'lazy', 'maybe_evaluate', 'head_from_fun',
    'maybe', 'fun_accepts_kwargs',
)

FUNHEAD_TEMPLATE = """
def {fun_name}({fun_args}):
    return {fun_value}
"""


class DummyContext:

    def __enter__(self):
        return self

    def __exit__(self, *exc_info):
        pass


[docs]class mlazy(lazy): """Memoized lazy evaluation. The function is only evaluated once, every subsequent access will return the same value. """ #: Set to :const:`True` after the object has been evaluated. evaluated = False _value = None
[docs] def evaluate(self): if not self.evaluated: self._value = super().evaluate() self.evaluated = True return self._value
[docs]def noop(*args, **kwargs): """No operation. Takes any arguments/keyword arguments and does nothing. """
def pass1(arg, *args, **kwargs): """Return the first positional argument.""" return arg def evaluate_promises(it): for value in it: if isinstance(value, promise): value = value() yield value
[docs]def first(predicate, it): """Return the first element in ``it`` that ``predicate`` accepts. If ``predicate`` is None it will return the first item that's not :const:`None`. """ return next( (v for v in evaluate_promises(it) if ( predicate(v) if predicate is not None else v is not None)), None, )
[docs]def firstmethod(method, on_call=None): """Multiple dispatch. Return a function that with a list of instances, finds the first instance that gives a value for the given method. The list can also contain lazy instances (:class:`~kombu.utils.functional.lazy`.) """ def _matcher(it, *args, **kwargs): for obj in it: try: meth = getattr(maybe_evaluate(obj), method) reply = (on_call(meth, *args, **kwargs) if on_call else meth(*args, **kwargs)) except AttributeError: pass else: if reply is not None: return reply return _matcher
[docs]def chunks(it, n): """Split an iterator into chunks with `n` elements each. Warning: ``it`` must be an actual iterator, if you pass this a concrete sequence will get you repeating elements. So ``chunks(iter(range(1000)), 10)`` is fine, but ``chunks(range(1000), 10)`` is not. Example: # n == 2 >>> x = chunks(iter([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]), 2) >>> list(x) [[0, 1], [2, 3], [4, 5], [6, 7], [8, 9], [10]] # n == 3 >>> x = chunks(iter([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]), 3) >>> list(x) [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9, 10]] """ for item in it: yield [item] + list(islice(it, n - 1))
[docs]def padlist(container, size, default=None): """Pad list with default elements. Example: >>> first, last, city = padlist(['George', 'Costanza', 'NYC'], 3) ('George', 'Costanza', 'NYC') >>> first, last, city = padlist(['George', 'Costanza'], 3) ('George', 'Costanza', None) >>> first, last, city, planet = padlist( ... ['George', 'Costanza', 'NYC'], 4, default='Earth', ... ) ('George', 'Costanza', 'NYC', 'Earth') """ return list(container)[:size] + [default] * (size - len(container))
[docs]def mattrgetter(*attrs): """Get attributes, ignoring attribute errors. Like :func:`operator.itemgetter` but return :const:`None` on missing attributes instead of raising :exc:`AttributeError`. """ return lambda obj: {attr: getattr(obj, attr, None) for attr in attrs}
[docs]def uniq(it): """Return all unique elements in ``it``, preserving order.""" seen = set() return (seen.add(obj) or obj for obj in it if obj not in seen)
def lookahead(it): """Yield pairs of (current, next) items in `it`. `next` is None if `current` is the last item. Example: >>> list(lookahead(x for x in range(6))) [(0, 1), (1, 2), (2, 3), (3, 4), (4, 5), (5, None)] """ a, b = tee(it) next(b, None) return zip_longest(a, b)
[docs]def regen(it): """Convert iterator to an object that can be consumed multiple times. ``Regen`` takes any iterable, and if the object is an generator it will cache the evaluated list on first access, so that the generator can be "consumed" multiple times. """ if isinstance(it, (list, tuple)): return it return _regen(it)
class _regen(UserList, list): # must be subclass of list so that json can encode. def __init__(self, it): # pylint: disable=super-init-not-called # UserList creates a new list and sets .data, so we don't # want to call init here. self.__it = it self.__consumed = [] self.__done = False def __reduce__(self): return list, (self.data,) def __length_hint__(self): return self.__it.__length_hint__() def __lookahead_consume(self, limit=None): if not self.__done and (limit is None or limit > 0): it = iter(self.__it) try: now = next(it) except StopIteration: return self.__consumed.append(now) # Maintain a single look-ahead to ensure we set `__done` when the # underlying iterator gets exhausted while not self.__done: try: next_ = next(it) self.__consumed.append(next_) except StopIteration: self.__done = True break finally: yield now now = next_ # We can break out when `limit` is exhausted if limit is not None: limit -= 1 if limit <= 0: break def __iter__(self): yield from self.__consumed yield from self.__lookahead_consume() def __getitem__(self, index): if index < 0: return self.data[index] # Consume elements up to the desired index prior to attempting to # access it from within `__consumed` consume_count = index - len(self.__consumed) + 1 for _ in self.__lookahead_consume(limit=consume_count): pass return self.__consumed[index] def __bool__(self): if len(self.__consumed): return True try: next(iter(self)) except StopIteration: return False else: return True @property def data(self): if not self.__done: self.__consumed.extend(self.__it) self.__done = True return self.__consumed def __repr__(self): return "<{}: [{}{}]>".format( self.__class__.__name__, ", ".join(repr(e) for e in self.__consumed), "..." if not self.__done else "", ) def _argsfromspec(spec, replace_defaults=True): if spec.defaults: split = len(spec.defaults) defaults = (list(range(len(spec.defaults))) if replace_defaults else spec.defaults) positional = spec.args[:-split] optional = list(zip(spec.args[-split:], defaults)) else: positional, optional = spec.args, [] varargs = spec.varargs varkw = spec.varkw if spec.kwonlydefaults: kwonlyargs = set(spec.kwonlyargs) - set(spec.kwonlydefaults.keys()) if replace_defaults: kwonlyargs_optional = [ (kw, i) for i, kw in enumerate(spec.kwonlydefaults.keys()) ] else: kwonlyargs_optional = list(spec.kwonlydefaults.items()) else: kwonlyargs, kwonlyargs_optional = spec.kwonlyargs, [] return ', '.join(filter(None, [ ', '.join(positional), ', '.join(f'{k}={v}' for k, v in optional), f'*{varargs}' if varargs else None, '*' if (kwonlyargs or kwonlyargs_optional) and not varargs else None, ', '.join(kwonlyargs) if kwonlyargs else None, ', '.join(f'{k}="{v}"' for k, v in kwonlyargs_optional), f'**{varkw}' if varkw else None, ]))
[docs]def head_from_fun(fun, bound=False, debug=False): """Generate signature function from actual function.""" # we could use inspect.Signature here, but that implementation # is very slow since it implements the argument checking # in pure-Python. Instead we use exec to create a new function # with an empty body, meaning it has the same performance as # as just calling a function. is_function = inspect.isfunction(fun) is_callable = hasattr(fun, '__call__') is_cython = fun.__class__.__name__ == 'cython_function_or_method' is_method = inspect.ismethod(fun) if not is_function and is_callable and not is_method and not is_cython: name, fun = fun.__class__.__name__, fun.__call__ else: name = fun.__name__ definition = FUNHEAD_TEMPLATE.format( fun_name=name, fun_args=_argsfromspec(inspect.getfullargspec(fun)), fun_value=1, ) if debug: # pragma: no cover print(definition, file=sys.stderr) namespace = {'__name__': fun.__module__} # pylint: disable=exec-used # Tasks are rarely, if ever, created at runtime - exec here is fine. exec(definition, namespace) result = namespace[name] result._source = definition if bound: return partial(result, object()) return result
def arity_greater(fun, n): argspec = inspect.getfullargspec(fun) return argspec.varargs or len(argspec.args) > n def fun_takes_argument(name, fun, position=None): spec = inspect.getfullargspec(fun) return ( spec.varkw or spec.varargs or (len(spec.args) >= position if position else name in spec.args) )
[docs]def fun_accepts_kwargs(fun): """Return true if function accepts arbitrary keyword arguments.""" return any( p for p in inspect.signature(fun).parameters.values() if p.kind == p.VAR_KEYWORD )
[docs]def maybe(typ, val): """Call typ on value if val is defined.""" return typ(val) if val is not None else val
def seq_concat_item(seq, item): """Return copy of sequence seq with item added. Returns: Sequence: if seq is a tuple, the result will be a tuple, otherwise it depends on the implementation of ``__add__``. """ return seq + (item,) if isinstance(seq, tuple) else seq + [item] def seq_concat_seq(a, b): """Concatenate two sequences: ``a + b``. Returns: Sequence: The return value will depend on the largest sequence - if b is larger and is a tuple, the return value will be a tuple. - if a is larger and is a list, the return value will be a list, """ # find the type of the largest sequence prefer = type(max([a, b], key=len)) # convert the smallest list to the type of the largest sequence. if not isinstance(a, prefer): a = prefer(a) if not isinstance(b, prefer): b = prefer(b) return a + b