Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@curry
@add_repr
def append_bytes(path: str, content: bytes) -> Effect[HasFiles, OSError, None]:
"""
Get an `Effect` that appends to a file
Example:
>>> class Env:
... files = Files()
>>> append_bytes('foo.txt')(b'content of foo.txt')\\
... .discard_and_then(read('foo.txt'))\\
... .run(Env())
'content of foo.txt'
Args:
path: the path of the file to be written
content the content to append
@curry
def map_m(f: Callable[[A], Either[B, C]],
iterable: Iterable[A]) -> Either[Iterable[B], C]:
"""
Map each in element in ``iterable`` to
an `Either` by applying ``f``,
combine the elements by ``and_then``
from left to right and collect the results
Example:
>>> map_m(Right, range(3))
Right((0, 1, 2))
Args:
f: Function to map over ``iterable``
iterable: Iterable to map ``f`` over
Return:
@curry
def as_type(type_: Type[T], results: Results) -> Try[TypeError, List[T]]:
"""
Convert database results to `type_`
Example:
>>> results = pfun.effect.success(
... pfun.List([pfun.Dict(dict(name='bob', age=32))])
... )
>>> class User(pfun.Immutable):
... name: str
... age: int
>>> results.and_then(as_type(User))(None)
List((User(name='bob', age=32),))
Args:
type_: type to convert database results to
@curry
@add_repr
def map_m(f: Callable[[A1], Effect[R1, E1, A1]],
iterable: Iterable[A1]) -> Effect[R1, E1, Iterable[A1]]:
"""
Map each in element in ``iterable`` to
an `Effect` by applying ``f``,
combine the elements by ``and_then``
from left to right and collect the results
Example:
>>> map_m(success, range(3)).run(None)
(0, 1, 2)
Args:
f: Function to map over ``iterable``
iterable: Iterable to map ``f`` over
@curry
def map_m_(
value: Callable[[Any], Monad],
f: Callable[[Any], Monad],
iterable: Iterable
) -> Monad:
m_iterable = (f(x) for x in iterable)
return sequence_(value, m_iterable)
@curry
@add_repr
def write_bytes(path: str, content: bytes) -> Effect[HasFiles, OSError, None]:
"""
Get an `Effect` that writes to a file
Example:
>>> class Env:
... files = Files()
>>> write_bytes('foo.txt')(b'content of foo.txt')\\
... .discard_and_then(read('foo.txt'))\\
... .run(Env())
'content of foo.txt'
Args:
path: the path of the file to be written
content the content to write
@curry
def filter_m(f: Callable[[A], Maybe[bool]],
iterable: Iterable[A]) -> Maybe[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: Just(v % 2 == 0), range(3))
Just((0, 2))
Args:
f: Function to map ``iterable`` by
iterable: Iterable to map by ``f``
Return:
`iterable` mapped and filtered by `f`
@curry
def filter_m_(
value: Callable[[Any], Monad],
f: Callable[[Any], Monad],
iterable: Iterable
) -> Monad:
def combine(ms, mbx):
mb, x = mbx
return ms.and_then(
lambda xs: mb.and_then(lambda b: value(xs + (x, ) if b else xs))
)
mbs = (f(x) for x in iterable)
mbxs = zip(mbs, iterable)
return reduce(combine, mbxs, value(()))
@curry
def filter_m(f: Callable[[A], List[bool]],
iterable: Iterable[A]) -> List[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: List([v % 2 == 0]), range(3))
List(((0, 2),))
Args:
f: Function to map ``iterable`` by
iterable: Iterable to map by ``f``
Return:
`iterable` mapped and filtered by `f`
"""
@curry
@add_repr
def write(path: str, content: str) -> Effect[HasFiles, OSError, None]:
"""
Get an `Effect` that writes to a file
Example:
>>> class Env:
... files = Files()
>>> write('foo.txt')('contents')\\
... .discard_and_then(read('foo.txt'))\\
... .run(Env())
'content of foo.txt'
Args:
path: the path of the file to be written
content the content to write