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
|
import collections
import functools
import itertools
from typing import ( # noqa: F401
Any,
Callable,
Dict,
Iterable,
List,
Mapping,
Set,
Tuple,
TypeVar,
Union,
)
from .toolz import (
compose as _compose,
)
T = TypeVar("T")
def identity(value: T) -> T:
return value
TGIn = TypeVar("TGIn")
TGOut = TypeVar("TGOut")
TFOut = TypeVar("TFOut")
def combine(
f: Callable[[TGOut], TFOut], g: Callable[[TGIn], TGOut]
) -> Callable[[TGIn], TFOut]:
return lambda x: f(g(x))
def apply_to_return_value(
callback: Callable[..., T]
) -> Callable[..., Callable[..., T]]:
def outer(fn: Callable[..., T]) -> Callable[..., T]:
# We would need to type annotate *args and **kwargs but doing so segfaults
# the PyPy builds. We ignore instead.
@functools.wraps(fn)
def inner(*args, **kwargs) -> T: # type: ignore
return callback(fn(*args, **kwargs))
return inner
return outer
TVal = TypeVar("TVal")
TKey = TypeVar("TKey")
to_tuple = apply_to_return_value(
tuple
) # type: Callable[[Callable[..., Iterable[TVal]]], Callable[..., Tuple[TVal, ...]]] # noqa: E501
to_list = apply_to_return_value(
list
) # type: Callable[[Callable[..., Iterable[TVal]]], Callable[..., List[TVal]]] # noqa: E501
to_set = apply_to_return_value(
set
) # type: Callable[[Callable[..., Iterable[TVal]]], Callable[..., Set[TVal]]] # noqa: E501
to_dict = apply_to_return_value(
dict
) # type: Callable[[Callable[..., Iterable[Union[Mapping[TKey, TVal], Tuple[TKey, TVal]]]]], Callable[..., Dict[TKey, TVal]]] # noqa: E501
to_ordered_dict = apply_to_return_value(
collections.OrderedDict
) # type: Callable[[Callable[..., Iterable[Union[Mapping[TKey, TVal], Tuple[TKey, TVal]]]]], Callable[..., collections.OrderedDict[TKey, TVal]]] # noqa: E501
sort_return = _compose(to_tuple, apply_to_return_value(sorted))
flatten_return = _compose(
to_tuple, apply_to_return_value(itertools.chain.from_iterable)
)
reversed_return = _compose(to_tuple, apply_to_return_value(reversed), to_tuple)
|