Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@curry
def map_m(f: Callable[[A], State[A, B]],
iterable: Iterable[A]) -> State[Iterable[A], B]:
"""
Map each in element in ``iterable`` to
an :class:`Maybe` by applying ``f``,
combine the elements by ``and_then``
from left to right and collect the results
:example:
>>> map_m(value, range(3)).run(None)
((0, 1, 2), None)
:param f: Function to map over ``iterable``
:param iterable: Iterable to map ``f`` over
:return: ``f`` mapped over ``iterable`` and combined from left to right.
"""
@curry
def filter_m(f: Callable[[A], State[bool, B]],
iterable: Iterable[A]) -> State[Iterable[A], B]:
"""
Map each element in ``iterable`` by applying ``f``,
filter the results by the value returned by ``f``
and combine from left to right.
:example:
>>> filter_m(lambda v: value(v % 2 == 0), range(3)).Run(None)
((0, 2), None)
:param f: Function to map ``iterable`` by
:param iterable: Iterable to map by ``f``
:return: `iterable` mapped and filtered by `f`
"""
return cast(State[Iterable[A], B], filter_m_(value, f, iterable))
@curry
def map_m(f: Callable[[A], Cont[A, B]],
iterable: Iterable[A]) -> Cont[Iterable[A], B]:
"""
Apply ``f`` to each element in ``iterable`` and collect the results
:example:
>>> from pfun import identity
>>> map_m(value, range(3)).run(identity)
(0, 1, 2)
:param f: The function to map over ``iterable``
:param iterable: The iterable to map over
:return: ``iterable`` mapped with ``f`` inside Cont
"""
return cast(Cont[Iterable[A], B], map_m_(value, f, iterable))
@curry
def with_effect_(
value: Callable[[Any], M], f: Callable[..., Generator[M, Any, Any]]
) -> Callable[..., M]:
@wraps(f)
def decorator(*args, **kwargs):
g = f(*args, **kwargs)
def cont(v) -> M:
try:
return g.send(v).and_then(cont)
except StopIteration as e:
return value(e.value)
try:
m = next(g)
return m.and_then(cont)
@curry
def map_m(f: Callable[[A], Reader[Context, B]],
iterable: Iterable[A]) -> Reader[Context, Iterable[B]]:
"""
Map each in element in ``iterable`` to
a :class:`Reader` by applying ``f``,
combine the elements by ``and_then``
from left to right and collect the results
:example:
>>> map_m(value, range(3)).run(...)
(0, 1, 2)
:param f: Function to map over ``iterable``
:param iterable: Iterable to map ``f`` over
:return: ``f`` mapped over ``iterable`` and combined from left to right.
"""
@curry
def map_m(f: Callable[[A], Writer[B, M]],
iterable: Iterable[A]) -> Writer[Iterable[B], M]:
"""
Map each in element in ``iterable`` to
an :class:`Writer` by applying ``f``,
combine the elements by ``and_then``
from left to right and collect the results
:example:
>>> map_m(value, range(3))
Writer((0, 1, 2), ...)
:param f: Function to map over ``iterable``
:param iterable: Iterable to map ``f`` over
:return: ``f`` mapped over ``iterable`` and combined from left to right.
"""
@curry
def filter_m(f: Callable[[A], IO[bool]],
iterable: Iterable[A]) -> IO[Iterable[A]]:
"""
Map each element in ``iterable`` by applying ``f``,
filter the results by the value returned by ``f``
and combine from left to right.
:example:
>>> filter_m(lambda v: IO(v % 2 == 0), range(3)).run()
(0, 2)
:param f: Function to map ``iterable`` by
:param iterable: Iterable to map by ``f``
:return: `iterable` mapped and filtered by `f`
"""
return cast(IO[Iterable[A]], filter_m_(value, f, iterable))
@curry
def write_str(path: str, content: str,
mode: Literal['w', 'a'] = 'w') -> IO[None]:
"""
Write a `str` to a file
:param path: the file to write to
:param content: the content to write to the file
:return: :class:`IO` action that produces \
the content of `filename` when run
"""
def run() -> Trampoline[None]:
with open(path, mode) as f:
f.write(content)
return Done(None)
return IO(run)
@curry
def write_bytes(path: str, content: bytes,
mode: Literal['w', 'a'] = 'w') -> IO[None]:
"""
Write `bytes` to a file
:param filename: the file to write to
:param content: the `bytes` to write to the file
:return: :class:`IO` action that writes to the file when run
"""
def run() -> Trampoline[None]:
with open(path, mode + 'b') as f:
f.write(content)
return Done(None)
return IO(run)
@curry
def map_m(f: Callable[[A], Free[F, B, C, D]],
iterable: Iterable[A]) -> Free[F, Iterable[B], C, D]:
"""
Map each in element in ``iterable`` to
a :class:`Free` by applying ``f``,
combine the elements by ``and_then``
from left to right and collect the results
:param f: Function to map over ``iterable``
:param iterable: Iterable to map ``f`` over
:return: ``f`` mapped over ``iterable`` and combined from left to right.
"""
return cast(Free[F, Iterable[B], C, D], map_m_(Done, f, iterable))