"""base classes and interfaces"""
import abc
import typing as t
from itertools import starmap
from types import GeneratorType
from .utils import CallableAsMethod
__all__ = [
"Generable",
"GeneratorCallable",
"ReusableGenerator",
]
T_yield = t.TypeVar("T_yield")
T_send = t.TypeVar("T_send")
T_return = t.TypeVar("T_return")
[docs]
class Generable(t.Generic[T_yield, T_send, T_return], t.Iterable[T_yield]):
"""ABC for generable objects.
Any object where :meth:`~object.__iter__`
returns a generator implements it.
"""
[docs]
@abc.abstractmethod
def __iter__(self):
"""
Returns
-------
~typing.Generator[T_yield, T_send, T_return]
the generator iterator
"""
raise NotImplementedError()
Generable.register(GeneratorType)
[docs]
class GeneratorCallable(t.Generic[T_yield, T_send, T_return]):
"""ABC for callables which return a generator.
Note that :term:`generator functions <generator>` already implement this.
"""
[docs]
def __call__(self, *args, **kwargs):
"""
Returns
-------
~typing.Generator[T_yield, T_send, T_return]
the resulting generator
"""
raise NotImplementedError()
class ReusableGeneratorMeta(CallableAsMethod, type(Generable)):
pass
[docs]
class ReusableGenerator(
Generable[T_yield, T_send, T_return], metaclass=ReusableGeneratorMeta
):
"""base class for reusable generator functions
Warning
-------
* Do not subclass directly. Subclasses are created with
the :func:`~gentools.core.reusable` decorator.
"""
[docs]
def __init__(self, *args, **kwargs):
self._bound_args = self.__signature__.bind(*args, **kwargs)
self._bound_args.apply_defaults()
[docs]
def __iter__(self):
return self.__wrapped__(
*self._bound_args.args, **self._bound_args.kwargs
)
[docs]
def __eq__(self, other):
if isinstance(other, self.__class__):
return self._bound_args.arguments == other._bound_args.arguments
return NotImplemented
[docs]
def __repr__(self):
fields = starmap("{}={!r}".format, self._bound_args.arguments.items())
return "{}({})".format(self.__class__.__qualname__, ", ".join(fields))
[docs]
def __hash__(self):
return hash(
(self._bound_args.args, tuple(self._bound_args.kwargs.items()))
)
[docs]
def replace(self, **kwargs):
"""create a new instance with certain fields replaced
Parameters
----------
**kwargs
fields to replace
Returns
-------
ReusableGenerator
a copy with replaced fields
"""
copied = self.__signature__.bind(
*self._bound_args.args, **self._bound_args.kwargs
)
copied.arguments.update(**kwargs)
return self.__class__(*copied.args, **copied.kwargs)