Source code for returns.transducers.transducers

from typing import Any, Callable, Iterable, Optional, TypeVar, overload

_ValueType = TypeVar('_ValueType')
_NewValueType = TypeVar('_NewValueType')

_AccValueType = TypeVar('_AccValueType')


[docs]class Reduced(object):
    """
    Sentinel for early termination inside transducer.

    .. code:: python

      >>> from returns.transducers import tmap, transduce, Reduced

      >>> def add_one(number: int) -> int:
      ...     return number + 1

      >>> def add(acc: int, number: int) -> int:
      ...     if acc == 3:
      ...         return Reduced(acc)
      ...     return acc + number

      >>> my_list = [0, 1, 2]
      >>> assert transduce(tmap(add_one), add, 0, my_list) == 3

    """

    _inner_value: Any

    def __init__(self, inner_value: Any) -> None:
        self._inner_value = inner_value

    @property
    def value(self) -> Any:
        return self._inner_value


class _Missing(object):
    """Represents a missing value for reducers."""

    _instance: Optional['_Missing'] = None

    def __new__(cls, *args: Any, **kwargs: Any) -> '_Missing':
        if cls._instance is None:
            cls._instance = object.__new__(cls)  # noqa: WPS609
        return cls._instance


#: A singleton representing any missing value
Missing = _Missing()


[docs]def transduce(
    xform: Callable[
        [Callable[[_AccValueType, _ValueType], _AccValueType]],
        Callable[[_AccValueType, _ValueType], _AccValueType],
    ],
    reducing_function: Callable[[_AccValueType, _ValueType], _AccValueType],
    initial: _AccValueType,
    iterable: Iterable[_ValueType],
) -> _AccValueType:
    """
    Process information with transducers.

    .. code:: python

      >>> from returns.transducers import tmap, transduce

      >>> def add_one(number: int) -> int:
      ...     return number + 1

      >>> def add(acc: int, number: int) -> int:
      ...     return acc + number

      >>> my_list = [0, 1, 2]
      >>> assert transduce(tmap(add_one), add, 0, my_list) == 6
    """
    reducer = xform(reducing_function)
    return reduce(reducer, iterable, initial)


@overload
def reduce(
    function: Callable[[_ValueType, _ValueType], _ValueType],
    iterable: Iterable[_ValueType],
    initial: _Missing = Missing,
) -> _ValueType:
    """Reduce without an initial value."""


@overload
def reduce(
    function: Callable[[_AccValueType, _ValueType], _AccValueType],
    iterable: Iterable[_ValueType],
    initial: _AccValueType,
) -> _AccValueType:
    """Reduce with an initial value."""


[docs]def reduce(function, iterable, initial=Missing):
    """
    A rewritten version of :func:`reduce <functools.reduce>`.

    This version considers some features borrowed from Clojure:

    - Early termination
    - Function initializer [TODO]

    You can use it as a normal reduce if you want:

    .. code:: python

      >>> from returns.transducers import reduce

      >>> def add(acc: int, value: int) -> int:
      ...     return acc + value

      >>> assert reduce(add, [1, 2, 3]) == 6

    """
    it = iter(iterable)

    if initial is Missing:
        try:
            acc_value = next(it)
        except StopIteration:
            raise TypeError(
                'reduce() of empty iterable with no initial value',
            ) from None
    else:
        acc_value = initial

    for value in it:  # noqa: WPS110
        acc_value = function(acc_value, value)
        if isinstance(acc_value, Reduced):
            return acc_value.value
    return acc_value