danom 0.5.0

Functional streams and monads

danom

API Reference

Ok

Frozen instance of an Ok monad used to wrap successful operations.

Ok.and_then

Ok.and_then(self, func: collections.abc.Callable[[~T], danom._result.Result], **kwargs: dict) -> danom._result.Result

Pipe another function that returns a monad.

>>> Ok(1).and_then(add_one) == Ok(2)
>>> Ok(1).and_then(raise_err) == Err(error=TypeError())

Ok.is_ok

Ok.is_ok(self) -> Literal[True]

Returns True if the result type is Ok.

>>> Ok().is_ok() == True

Ok.match

Ok.match(self, if_ok_func: collections.abc.Callable[[~T], danom._result.Result], _if_err_func: collections.abc.Callable[[~T], danom._result.Result]) -> danom._result.Result

Map Ok func to Ok and Err func to Err

>>> Ok(1).match(add_one, mock_get_error_type) == Ok(inner=2)
>>> Ok("ok").match(double, mock_get_error_type) == Ok(inner='okok')
>>> Err(error=TypeError()).match(double, mock_get_error_type) == Ok(inner='TypeError')

Ok.unwrap

Ok.unwrap(self) -> ~T

Unwrap the Ok monad and get the inner value.

>>> Ok().unwrap() == None
>>> Ok(1).unwrap() == 1
>>> Ok("ok").unwrap() == 'ok'

Err

Frozen instance of an Err monad used to wrap failed operations.

Err.and_then

Err.and_then(self, _: 'Callable[[T], Result]', **_kwargs: 'dict') -> 'Self'

Pipe another function that returns a monad. For Err will return original error.

>>> Err(error=TypeError()).and_then(add_one) == Err(error=TypeError())
>>> Err(error=TypeError()).and_then(raise_value_err) == Err(error=TypeError())

Err.is_ok

Err.is_ok(self) -> 'Literal[False]'

Returns False if the result type is Err.

Err().is_ok() == False

Err.match

Err.match(self, _if_ok_func: 'Callable[[T], Result]', if_err_func: 'Callable[[T], Result]') -> 'Result'

Map Ok func to Ok and Err func to Err

>>> Ok(1).match(add_one, mock_get_error_type) == Ok(inner=2)
>>> Ok("ok").match(double, mock_get_error_type) == Ok(inner='okok')
>>> Err(error=TypeError()).match(double, mock_get_error_type) == Ok(inner='TypeError')

Err.unwrap

Err.unwrap(self) -> 'None'

Unwrap the Err monad will raise the inner error.

>>> Err(error=TypeError()).unwrap() raise TypeError(...)

Stream

A lazy iterator with functional operations.

Stream.collect

Stream.collect(self) -> 'tuple'

Materialise the sequence from the Stream.

>>> stream = Stream.from_iterable([0, 1, 2, 3]).map(add_one)
>>> stream.collect() == (1, 2, 3, 4)

Stream.filter

Stream.filter(self, *fns: 'Callable[[T], bool]') -> 'Self'

Filter the stream based on a predicate. Will return a new Stream with the modified sequence.

>>> Stream.from_iterable([0, 1, 2, 3]).filter(lambda x: x % 2 == 0).collect() == (0, 2)

Simple functions can be passed in sequence to compose more complex filters

>>> Stream.from_iterable(range(20)).filter(divisible_by_3, divisible_by_5).collect() == (0, 15)

Stream.from_iterable

Stream.from_iterable(it: 'Iterable') -> 'Self'

This is the recommended way of creating a Stream object.

>>> Stream.from_iterable([0, 1, 2, 3]).collect() == (0, 1, 2, 3)

Stream.map

Stream.map(self, *fns: 'Callable[[T], U]') -> 'Self'

Map a function to the elements in the Stream. Will return a new Stream with the modified sequence.

>>> Stream.from_iterable([0, 1, 2, 3]).map(add_one).collect() == (1, 2, 3, 4)

This can also be mixed with safe functions:

>>> Stream.from_iterable([0, 1, 2, 3]).map(add_one).collect() == (Ok(inner=1), Ok(inner=2), Ok(inner=3), Ok(inner=4))

>>> @safe
... def two_div_value(x: float) -> float:
...     return 2 / x

>>> Stream.from_iterable([0, 1, 2, 4]).map(two_div_value).collect() == (Err(error=ZeroDivisionError('division by zero')), Ok(inner=2.0), Ok(inner=1.0), Ok(inner=0.5))

Simple functions can be passed in sequence to compose more complex transformations

>>> Stream.from_iterable(range(5)).map(mul_two, add_one).collect() == (1, 3, 5, 7, 9)

Stream.par_collect

Stream.par_collect(self, workers: 'int' = 4, *, use_threads: 'bool' = False) -> 'tuple'

Materialise the sequence from the Stream in parallel.

>>> stream = Stream.from_iterable([0, 1, 2, 3]).map(add_one)
>>> stream.par_collect() == (1, 2, 3, 4)

Use the workers arg to select the number of workers to use. Use -1 to use all available processors (except 1). Defaults to 4.

>>> stream = Stream.from_iterable([0, 1, 2, 3]).map(add_one)
>>> stream.par_collect(workers=-1) == (1, 2, 3, 4)

For smaller I/O bound tasks use the use_threads flag as True. If False the processing will use ProcessPoolExecutor else it will use ThreadPoolExecutor.

>>> stream = Stream.from_iterable([0, 1, 2, 3]).map(add_one)
>>> stream.par_collect(use_threads=True) == (1, 2, 3, 4)

Stream.partition

Stream.partition(self, fn: 'Callable[[T], bool]', *, workers: 'int' = 1, use_threads: 'bool' = False) -> 'tuple[Self, Self]'

Similar to filter except splits the True and False values. Will return a two new Stream with the partitioned sequences.

Each partition is independently replayable.

>>> part1, part2 = Stream.from_iterable([0, 1, 2, 3]).partition(lambda x: x % 2 == 0)
>>> part1.collect() == (0, 2)
>>> part2.collect() == (1, 3)

As partition triggers an action, the parameters will be forwarded to the collect call if the workers are greater than 1.

>>> Stream.from_iterable(range(10)).map(add_one, add_one).partition(divisible_by_3, workers=4)
>>> part1.map(add_one).par_collect() == (4, 7, 10)
>>> part2.collect() == (2, 4, 5, 7, 8, 10, 11)

safe

safe

safe(func: collections.abc.Callable[~P, ~T]) -> collections.abc.Callable[~P, danom._result.Result]

Decorator for functions that wraps the function in a try except returns Ok on success else Err.

>>> @safe
... def add_one(a: int) -> int:
...     return a + 1

>>> add_one(1) == Ok(inner=2)

safe_method

safe_method

safe_method(func: collections.abc.Callable[~P, ~T]) -> collections.abc.Callable[~P, danom._result.Result]

The same as safe except it forwards on the self of the class instance to the wrapped function.

>>> class Adder:
...     def __init__(self, result: int = 0) -> None:
...         self.result = result
...
...     @safe_method
...     def add_one(self, a: int) -> int:
...         return self.result + 1

>>> Adder.add_one(1) == Ok(inner=1)

compose

compose

compose(*fns: collections.abc.Callable[[T], U]) -> collections.abc.Callable[[T], U]

Compose multiple functions into one.

The functions will be called in sequence with the result of one being used as the input for the next.

>>> add_two = compose(add_one, add_one)
>>> add_two(0) == 2

>>> add_two = compose(add_one, add_one, is_even)
>>> add_two(0) == True

all_of

all_of

all_of(*fns: collections.abc.Callable[[T], bool]) -> collections.abc.Callable[[T], bool]

True if all of the given functions return True.

>>> is_valid_user = all_of(is_subscribed, is_active, has_2fa)
>>> is_valid_user(user) == True

any_of

any_of

any_of(*fns: collections.abc.Callable[[T], bool]) -> collections.abc.Callable[[T], bool]

True if any of the given functions return True.

>>> is_eligible = any_of(has_coupon, is_vip, is_staff)
>>> is_eligible(user) == True

identity

identity

identity(x: T) -> T

Basic identity function.

>>> identity("abc") == "abc"
>>> identity(1) == 1
>>> identity(ComplexDataType(a=1, b=2, c=3)) == ComplexDataType(a=1, b=2, c=3)

invert

invert

invert(func: collections.abc.Callable[~P, bool]) -> collections.abc.Callable[~P, bool]

Invert a boolean function so it returns False where it would've returned True.

>>> invert(has_len)("abc") == False
>>> invert(has_len)("") == True

new_type

new_type

new_type(name: 'str', base_type: 'type', validators: 'Callable | Sequence[Callable] | None' = None, converters: 'Callable | Sequence[Callable] | None' = None, *, frozen: 'bool' = True)

Create a NewType based on another type.

>>> def is_positive(value):
...     return value >= 0

>>> ValidBalance = new_type("ValidBalance", float, validators=[is_positive])
>>> ValidBalance("20") == ValidBalance(inner=20.0)

Unlike an inherited class, the type will not return True for an isinstance check.

>>> isinstance(ValidBalance(20.0), ValidBalance) == True
>>> isinstance(ValidBalance(20.0), float) == False

The methods of the given base_type will be forwarded to the specialised type. Alternatively the map method can be used to return a new type instance with the transformation.

>>> def has_len(email: str) -> bool:
... return len(email) > 0

>>> Email = new_type("Email", str, validators=[has_len])
>>> Email("[e-mail]").upper() == "[e-mail]"
>>> Email("[e-mail]").map(str.upper) == Email(inner='[e-mail]')

::

Repo map

├── .github
│   └── workflows
│       ├── ci_tests.yaml
│       └── publish.yaml
├── dev_tools
│   ├── __init__.py
│   ├── update_cov.py
│   └── update_readme.py
├── src
│   └── danom
│       ├── __init__.py
│       ├── _err.py
│       ├── _new_type.py
│       ├── _ok.py
│       ├── _result.py
│       ├── _safe.py
│       ├── _stream.py
│       └── _utils.py
├── tests
│   ├── __init__.py
│   ├── conftest.py
│   ├── test_api.py
│   ├── test_err.py
│   ├── test_new_type.py
│   ├── test_ok.py
│   ├── test_result.py
│   ├── test_safe.py
│   ├── test_stream.py
│   └── test_utils.py
├── .pre-commit-config.yaml
├── README.md
├── pyproject.toml
├── ruff.toml
└── uv.lock
::