betty.functools module#

Provide functional programming utilities.

class betty.functools.CallableDecorator#

Bases: Generic[P, ReturnT, DecoratedP, DecoratedReturnT]

An object capable of decorating a callable.

__init__(*, callable_decorator: DecoratorCallableType[P, ReturnT, DecoratedP, DecoratedReturnT])#
final class betty.functools.DecoratedCallable#

Bases: Generic[P, ReturnT]

Apply a decorator to a callable.

__init__(decorator: DecoratorCallableType[P, ReturnT, DecoratedP, DecoratedReturnT], decorated: DecoratedCallableType[DecoratedP, DecoratedReturnT], /)#
class betty.functools.Do#

Bases: Generic[DoFP, DoFReturnT]

A functional implementation of do-while functionality, with retries and timeouts.

__init__(do: Callable[DoFP, DoFReturnT | Awaitable[DoFReturnT]], *do_args: DoFP.args, **do_kwargs: DoFP.kwargs)#
async until(*conditions: Callable[[DoFReturnT], None | bool | Awaitable[None | bool]], retries: int = 5, timeout: int = 300, interval: float = 0.1) → DoFReturnT#

Perform the ‘do’ until it succeeds or as long as the given arguments allow.

Parameters:

  • timeout – The timeout in seconds.

  • interval – The interval between ‘loops’ in seconds.

final class betty.functools.LazyReCallable#

Bases: Generic

A value that can be called multiple times while always returning the exact same value.

The proxied callable will at most be called once.

__init__(factory: Callable[[], ValueT], /)#
final class betty.functools.Pipeline#

Bases: Generic

A function pipeline.

Function pipeline let you pipe/chain/link/combine functions into pipelines that take an input value and, if the functions pass, return an output value. Each pipeline may be (re)used as many times as needed.

__init__(pipe: Pipe[ValueT, ReturnT], /)#
pipe(pipe: Pipe[ReturnT, PipeReturnT], /) → Pipeline[ValueT, PipeReturnT]#

Return a new pipeline consisting of self with pipe added to it.

final class betty.functools.Result#

Bases: Generic[P, T]

Decorate a callable and store its return value or raised exception.

__init__(target: Callable[P, T], /)#
result() → T#

Get the target’s return value.

If the target raised an exception, calling this method will re-raise the exception.

exception betty.functools.ResultUnavailable#

Bases: RuntimeError

A betty.functools.Result.result is unavailable.

__init__()#
betty.functools.map_suppress(raising_map: Callable[[T], U], exception_type: type[BaseException], items: Iterable[T], /) → Iterator[U]#

Map values, skipping those for which the application of raising_map raises errors.

betty.functools.passthrough(value: T, /) → T#

Return the value.

betty.functools.suppress(target: Callable[P, T], *exceptions: type[BaseException]) → Callable[P, T | type[Nothing]]#

Return the value, but suppress any errors.

betty.functools.unique(*values: Iterable[ValueT], key: Callable[[ValueT], Any] | None = None) → Iterator[ValueT]#

Yield the first occurrences of values in a sequence.

For the purpose of filtering duplicate values from an iterable, this works similar to set, except that this class supports non-hashable values. It is therefore slightly slower than set.