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.

This is thread-safe, which means you can safely use this between different threads.

__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[Void]]

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.