Skip to content

pfun.effect.Effect dataclass

Wrapper for functions of type Callable[[R], Awaitable[pfun.Either[E, A]]] that are allowed to perform side-effects

__call__(self, r, max_processes=None, max_threads=None) special

Run the function wrapped by this Effect asynchronously, including potential side-effects. If the function fails the resulting error will be raised as an exception.

Parameters:

Name Type Description Default
r R

The dependency with which to run this Effect

 required
max_processes int

The max number of processes used to run cpu bound parts of this effect

 None
max_threads int

The max number of threads used to run io bound parts of this effect

 None

Returns:

Type Description
A

The succesful result of the wrapped function if it succeeds

Exceptions:

Type Description
E

If the Effect fails and E is a subclass of Exception
RuntimeError

if the effect fails and E is not a subclass of Exception

and_then(self, f)

Create new Effect that applies f to the result of running this effect successfully. If this Effect fails, f is not applied.

Examples:

1
2
 
>>> success(2).and_then(lambda i: success(i + 2)).run(None)
4

Parameters:

Name Type Description Default
f Callable[[A], Union[Awaitable[Effect[Any, E2, B]], Effect[Any, E2, B]]]

Function to pass the result of this Effect instance once it can be computed

 required

Returns:

Type Description
Effect[Any, Union[E, E2], B]

New Effect which wraps the result of passing the result of this Effect instance to f

discard_and_then(self, effect)

Create a new effect that discards the result of this effect, and produces instead effect. Like and_then but does not require you to handle the result. Convenient for effects that produce None, like writing to files.

Examples:

1
2
3
4
5
6
7
 
>>> from pfun import files
>>> class Env:
...     files = files.Files()
>>> files.write('foo.txt', 'Hello!')\
...     .discard_and_then(files.read('foo.txt'))\
...     .run(Env())
Hello!

Parameters:

Name Type Description Default
effect Effect[Any, E2, B]

Effect instance to run after this Effect has run successfully.

 required

Returns:

Type Description
Effect[Any, Union[E, E2], B]

New effect that succeeds with effect

either(self)

Push the potential error into the success channel as an either, allowing error handling.

Examples:

1
2
3
4
5
 
>>> error('Whoops!').either().map(
...     lambda either: either.get if isinstance(either, Right)
...                    else 'Phew!'
... ).run(None)
'Phew!'

Returns:

Type Description
Effect[R, NoReturn, Either[E, A]]

New Effect that produces a Left[E] if it has failed, or a :Right[A] if it succeeds

ensure(self, effect)

Create an Effect that will always run effect, regardless of whether this Effect succeeds or fails. The result of effect is ignored, and the resulting effect instead succeeds or fails with the succes or error value of this effect. Useful for closing resources.

Examples:

1
2
3
4
5
6
7
8
9
 
>>> from pfun.effect.console import Console
>>> console = Console()
>>> finalizer = console.print('finalizing!')
>>> success('result').ensure(finalizer).run(None)
finalizing!
'result'
>>> error('whoops!').ensure(finalizer).run(None)
finalizing!
RuntimeError: whoops!

Parameters:

Name Type Description Default
effect Effect[Any, NoReturn, Any]

Effect to run after this effect terminates either successfully or with an error

 required

Returns:

Type Description
Effect[Any, E, A]

Effect that fails or succeeds with the result of this effect, but always runs effect

map(self, f)

Map f over the result produced by this Effect once it is run

Examples:

1
2
 
>>> success(2).map(lambda v: v + 2).run(None)
4

Parameters:

Name Type Description Default
f Callable[[A], Union[Awaitable[B], B]]

function to map over this Effect

 required

Returns:

Type Description
Effect[R, E, B]

new Effect with f applied to the value produced by this Effect.

memoize(self)

Create an Effect that caches its result. When the effect is evaluated for the second time, its side-effects are not performed, it simply succeeds with the cached result. This means you should be careful with memoizing complicated effects. Useful for effects that have expensive results, such as calling a slow HTTP api or reading a large file.

Examples:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
 
>>> from pfun.console import Console
>>> console = Console()
>>> effect = console.print(
...     'Doing something expensive'
... ).discard_and_then(
...     success('result')
... ).memoize()
>>> # this would normally cause an effect to be run twice.
>>> double_effect = effect.discard_and_then(effect)
>>> double_effect.run(None)
Doing something expensive
'result'

Returns:

Type Description
Effect[R, E, A]

memoized Effect

recover(self, f)

Create new Effect that applies f to the error result of running this effect if it fails. If this Effect succeeds, f is not applied.

Examples:

1
2
 
>>> error('Whoops!').recover(lambda _: success('Phew!')).run(None)
'Phew!'

Parameters:

Name Type Description Default
f Callable[[E], Effect[Any, E2, B]]

Function to pass the error result of this Effect instance once it can be computed

 required

Returns:

Type Description
Effect[Any, E2, Union[A, B]]

New :Effect which wraps the result of passing the error result of this Effect instance to f

run(self, r, asyncio_run=<function run at 0x7fb19a737830>, max_processes=None, max_threads=None)

Run the function wrapped by this Effect, including potential side-effects. If the function fails the resulting error will be raised as an exception.

Parameters:

Name Type Description Default
r R

The dependency with which to run this Effect asyncio_run: Function to run the coroutine returned by the wrapped function

 required
max_processes int

The max number of processes used to run cpu bound parts of this effect

 None
max_threads int

The max number of threads used to run io bound parts of this effect

 None

Returns:

Type Description
A

The succesful result of the wrapped function if it succeeds

Exceptions:

Type Description
E

If the Effect fails and E is a subclass of Exception
RuntimeError

if the effect fails and E is not a subclass of Exception

pfun.effect.Success

Type-alias for Effect[object, NoReturn, TypeVar('A')].

pfun.effect.Try

Type-alias for Effect[object, TypeVar('E'), TypeVar('A')].

pfun.effect.Depends

Type-alias for Effect[TypeVar('R'), NoReturn, TypeVar('A')].

pfun.effect.success(value)

Wrap a function in Effect that does nothing but return value

Examples:

1
2
 
>>> success('Yay!').run(None)
'Yay!'

Parameters:

Name Type Description Default
value A1

The value to return when the Effect is executed

 required

Returns:

Type Description
Effect[object, NoReturn, A1]

Effect that wraps a function returning value

pfun.effect.error(reason)

Create an Effect that does nothing but fail with reason

Examples:

1
2
 
>>> error('Whoops!').run(None)
RuntimeError: 'Whoops!'

Parameters:

Name Type Description Default
reason E1

Value to fail with

 required

Returns:

Type Description
Effect[object, E1, NoReturn]

Effect that fails with reason

pfun.effect.depend(r_type=None)

Get an Effect that produces the dependency passed to run when executed

Examples:

1
2
 
>>> depend(str).run('dependency')
'dependency'

Parameters:

Name Type Description Default
r_type Optional[Type[R1]]

The expected dependency type of the resulting effect. Used ONLY for type-checking and doesn't impact runtime behaviour in any way

 None

Returns:

Type Description
Depends[R1, R1]

Effect that produces the dependency passed to run

pfun.effect.sequence_async(iterable)

Evaluate each Effect in iterable asynchronously and collect the results

Examples:

1
2
 
>>> sequence_async([success(v) for v in range(3)]).run(None)
(0, 1, 2)

Parameters:

Name Type Description Default
iterable Iterable[Effect[R1, E1, A1]]

The iterable to collect results from

 required

Returns:

Type Description
Effect[R1, E1, Iterable[A1]]

Effect that produces collected results

pfun.effect.filter_(f, iterable)

Map each element in iterable by applying f, filter the results by the value returned by f and combine from left to right.

Examples:

1
2
 
>>> filter(lambda v: success(v % 2 == 0), range(3)).run(None)
(0, 2)

Parameters:

Name Type Description Default
f Callable[[A], Effect[R1, E1, bool]]

Function to map iterable by

 required
iterable Iterable[A]

Iterable to map by f

 required

Returns:

Type Description
Effect[R1, E1, Iterable[A]]

iterable mapped and filtered by f

pfun.effect.for_each(f, iterable)

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

Examples:

1
2
 
>>> for_each(success, range(3)).run(None)
(0, 1, 2)

Parameters:

Name Type Description Default
f Callable[[A1], Effect[R1, E1, B]]

Function to map over iterable

 required
iterable Iterable[A1]

Iterable to map f over

 required

Returns:

Type Description
Effect[R1, E1, Iterable[B]]

f mapped over iterable and combined from left to right.

pfun.effect.absolve(effect)

Move the error type from an Effect producing an Either into the error channel of the Effect

Examples:

1
2
3
4
5
6
 
>>> effect = error('Whoops').either().map(
...     lambda either: either.get if isinstance(either, Right)
...                    else 'Phew!'
... )
>>> absolve(effect).run(None)
'Phew!'

Parameters:

Name Type Description Default
effect Effect[R1, NoReturn, Either[E1, A1]]

an Effect producing an Either

 required

Returns:

Type Description
Effect[R1, E1, A1]

an Effect failing with E1 or succeeding with A1

pfun.effect.combine(*effects)

Create an effect that produces the result of calling the passed function with the results of effects in effects

Examples:

1
2
 
>>> combine(success(2), success(2))(lambda a, b: a + b).run(None)
4

Parameters:

Name Type Description Default
effects Effect[R1, E1, A2]

Effects the results of which to pass to the combiner function

 ()

Returns:

Type Description
Callable[[Callable[..., Union[Awaitable[A1], A1]]], Effect[Any, Any, A1]]

function that takes a combiner function and returns an Effect that applies the function to the results of effects

pfun.effect.lift

Decorator that enables decorated functions to operate on Effect instances.

Examples:

1
2
3
4
 
>>> def add(a: int, b: int) -> int:
...     return a + b
>>> lift(add)(success(2), success(2)).run(None)
4

__call__(self, *effects) special

Parameters:

Name Type Description Default
args

Effect instances, the result of which should be passed to f

 required

Returns:

Type Description
Effect

f applied to the success values of args

__init__(self, f) special

Parameters:

Name Type Description Default
f L

The function to decorate

 required

pfun.effect.catch dataclass

Decorator that catches errors as an Effect. If the decorated function performs additional side-effects, they are not carried out until the effect is run.

Examples:

1
2
3
4
5
 
>>> f = catch(ZeroDivisionError)(lambda v: 1 / v)
>>> f(1).run(None)
1.0
>>> f(0).run(None)
ZeroDivisionError

__call__(self, f) special

Decorate f to catch exceptions as an Effect

__init__(self, error, *errors) special

Parameters:

Name Type Description Default
error Type[EX]

The first exception type to catch

 required
errors Type[EX]

The remaining exception types to catch

 ()

pfun.effect.from_awaitable(awaitable)

Create an Effect that produces the result of awaiting awaitable

Examples:

1
2
3
4
 
>>> async def f() -> str:
...     return 'Yay!'
>>> from_awaitable(f()).run(None)
'Yay'

Parameters:

Name Type Description Default
awaitable Awaitable[A1]

Awaitable to await in the resulting Effect

 required

Returns:

Type Description
Effect[object, NoReturn, A1]

Effect that produces the result of awaiting awaitable

pfun.effect.from_callable(f)

Create an Effect from a function that takes a dependency and returns an Either

Examples:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
 
>>> from pfun.either import Either, Left, Right
>>> def f(r: str) -> Either[str, str]:
...     if not r:
...         return Left('Empty string')
...     return Right(r * 2)
>>> effect = from_callable(f)
>>> effect.run('')
RuntimeError: Empty string
>>> effect.run('Hello!')
Hello!Hello!

Parameters:

Name Type Description Default
f Callable[[R1], Union[Awaitable[Either[E1, A1]], Either[E1, A1]]]

the function to turn into an Effect

 required

Returns:

Type Description
Effect[R1, E1, A1]

f as an Effect

pfun.effect.cpu_bound(f)

Decorator for functions that should be executed in separate a separate process during effect evaluation

Examples:

1
2
 
>>> success(2).map(cpu_bound(lambda v: v + 2)).run(None)
4

Parameters:

Name Type Description Default
f F1

The function to decorate

 required

Returns:

Type Description
F1

f decorated

pfun.effect.io_bound(f)

Decorator for functions that should be executed in separate a separate thread during effect evaluation

Examples:

1
2
 
>>> success(2).map(io_bound(lambda v: v + 2)).run(None)
4

Parameters:

Name Type Description Default
f F1

The function to decorate

 required

Returns:

Type Description
F1

f decorated