"""
Method chaining interface.
.. versionadded:: 1.0.0
"""
import typing as t
import pydash as pyd
from pydash.exceptions import InvalidMethod
from ..helpers import UNSET, Unset
from .all_funcs import AllFuncs
__all__ = (
"chain",
"tap",
)
ValueT_co = t.TypeVar("ValueT_co", covariant=True)
T = t.TypeVar("T")
T2 = t.TypeVar("T2")
class Chain(AllFuncs, t.Generic[ValueT_co]):
"""Enables chaining of :attr:`module` functions."""
#: Object that contains attribute references to available methods.
module = pyd
invalid_method_exception = InvalidMethod
def __init__(self, value: t.Union[ValueT_co, Unset] = UNSET) -> None:
self._value = value
def _wrap(self, func) -> "ChainWrapper[t.Union[ValueT_co, Unset]]":
"""Implement `AllFuncs` interface."""
return ChainWrapper(self._value, func)
def value(self) -> ValueT_co:
"""
Return current value of the chain operations.
Returns:
Current value of chain operations.
"""
return self(self._value)
def to_string(self) -> str:
"""
Return current value as string.
Returns:
Current value of chain operations casted to ``str``.
"""
return self.module.to_string(self.value())
def commit(self) -> "Chain[ValueT_co]":
"""
Executes the chained sequence and returns the wrapped result.
Returns:
New instance of :class:`Chain` with resolved value from
previous :class:`Class`.
"""
return Chain(self.value())
def plant(self, value: t.Any) -> "Chain[ValueT_co]":
"""
Return a clone of the chained sequence planting `value` as the wrapped value.
Args:
value: Value to plant as the initial chain value.
"""
# pylint: disable=no-member,maybe-no-member
wrapper = self._value
wrappers = []
if hasattr(wrapper, "_value"):
wrappers = [wrapper]
while isinstance(wrapper._value, ChainWrapper):
wrapper = wrapper._value # type: ignore
wrappers.insert(0, wrapper)
clone: Chain[t.Any] = Chain(value)
for wrap in wrappers:
clone = ChainWrapper(clone._value, wrap.method)( # type: ignore
*wrap.args, # type: ignore
**wrap.kwargs, # type: ignore
)
return clone
def __call__(self, value) -> ValueT_co:
"""
Return result of passing `value` through chained methods.
Args:
value: Initial value to pass through chained methods.
Returns:
Result of method chain evaluation of `value`.
"""
if isinstance(self._value, ChainWrapper):
# pylint: disable=maybe-no-member
value = self._value.unwrap(value)
return value
class ChainWrapper(t.Generic[ValueT_co]):
"""Wrap :class:`Chain` method call within a :class:`ChainWrapper` context."""
def __init__(self, value: ValueT_co, method) -> None:
self._value = value
self.method = method
self.args = ()
self.kwargs: t.Dict[t.Any, t.Any] = {}
def _generate(self):
"""Generate a copy of this instance."""
# pylint: disable=attribute-defined-outside-init
new = self.__class__.__new__(self.__class__)
new.__dict__ = self.__dict__.copy()
return new
def unwrap(self, value=UNSET):
"""
Execute :meth:`method` with :attr:`_value`, :attr:`args`, and :attr:`kwargs`.
If :attr:`_value` is an instance of :class:`ChainWrapper`, then unwrap it before calling
:attr:`method`.
"""
# Generate a copy of ourself so that we don't modify the chain wrapper
# _value directly. This way if we are late passing a value, we don't
# "freeze" the chain wrapper value when a value is first passed.
# Otherwise, we'd locked the chain wrapper value permanently and not be
# able to reuse it.
wrapper = self._generate()
if isinstance(wrapper._value, ChainWrapper):
# pylint: disable=no-member,maybe-no-member
wrapper._value = wrapper._value.unwrap(value)
elif not isinstance(value, ChainWrapper) and value is not UNSET:
# Override wrapper's initial value.
wrapper._value = value
if wrapper._value is not UNSET:
value = wrapper._value
return wrapper.method(value, *wrapper.args, **wrapper.kwargs)
def __call__(self, *args, **kwargs):
"""
Invoke the :attr:`method` with :attr:`value` as the first argument and return a new
:class:`Chain` object with the return value.
Returns:
New instance of :class:`Chain` with the results of :attr:`method` passed in as
value.
"""
self.args = args
self.kwargs = kwargs
return Chain(self)
class _Dash(object):
"""Class that provides attribute access to valid :mod:`pydash` methods and callable access to
:mod:`pydash` method chaining."""
def __getattr__(self, attr):
"""Proxy to :meth:`Chain.get_method`."""
return Chain.get_method(attr)
def __call__(self, value: t.Union[ValueT_co, Unset] = UNSET) -> Chain[ValueT_co]:
"""Return a new instance of :class:`Chain` with `value` as the seed."""
return Chain(value)
def chain(value: t.Union[T, Unset] = UNSET) -> Chain[T]:
"""
Creates a :class:`Chain` object which wraps the given value to enable intuitive method chaining.
Chaining is lazy and won't compute a final value until :meth:`Chain.value` is called.
Args:
value: Value to initialize chain operations with.
Returns:
Instance of :class:`Chain` initialized with `value`.
Example:
>>> chain([1, 2, 3, 4]).map(lambda x: x * 2).sum().value()
20
>>> chain().map(lambda x: x * 2).sum()([1, 2, 3, 4])
20
>>> summer = chain([1, 2, 3, 4]).sum()
>>> new_summer = summer.plant([1, 2])
>>> new_summer.value()
3
>>> summer.value()
10
>>> def echo(item):
... print(item)
>>> summer = chain([1, 2, 3, 4]).for_each(echo).sum()
>>> committed = summer.commit()
1
2
3
4
>>> committed.value()
10
>>> summer.value()
1
2
3
4
10
.. versionadded:: 1.0.0
.. versionchanged:: 2.0.0
Made chaining lazy.
.. versionchanged:: 3.0.0
- Added support for late passing of `value`.
- Added :meth:`Chain.plant` for replacing initial chain value.
- Added :meth:`Chain.commit` for returning a new :class:`Chain` instance initialized with
the results from calling :meth:`Chain.value`.
"""
return Chain(value)
def tap(value: T, interceptor: t.Callable[[T], t.Any]) -> T:
"""
Invokes `interceptor` with the `value` as the first argument and then returns `value`. The
purpose of this method is to "tap into" a method chain in order to perform operations on
intermediate results within the chain.
Args:
value: Current value of chain operation.
interceptor: Function called on `value`.
Returns:
`value` after `interceptor` call.
Example:
>>> data = []
>>> def log(value):
... data.append(value)
>>> chain([1, 2, 3, 4]).map(lambda x: x * 2).tap(log).value()
[2, 4, 6, 8]
>>> data
[[2, 4, 6, 8]]
.. versionadded:: 1.0.0
"""
interceptor(value)
return value
