danom documentation

class danom.AsyncStream(seq: tuple, ops: tuple = ())

Bases: _BaseAsyncStream, Generic

A stream that applies async functions to its values.

Version changes

0.16.0: Added AsyncStream

async collect(*, workers: int = 4, use_threads: bool = False) → tuple[U, ...]

Materialise the AsyncStream into a tuple.

values = await stream.collect()

filter(fn: ~collections.abc.Callable[[~P], ~collections.abc.Awaitable[bool]], *args: ~typing.~P, **kwargs: ~typing.~P) → AsyncStream[T]

Filter the AsyncStream with an async predicate.

await stream.filter(is_even).collect()

async fold(initial: T, fn: Callable[[T, U], T], *, workers: int = 1, use_threads: bool = False) → T

Reduce the collected values into one value.

total = await stream.fold(0, add)

classmethod from_iterable(it: Iterable) → Self

Create an AsyncStream from an iterable.

from danom import AsyncStream

stream = AsyncStream.from_iterable([1, 2, 3])

map(fn: ~collections.abc.Callable[[~P], ~collections.abc.Awaitable[~danom._stream._base.U]], *args: ~typing.~P, **kwargs: ~typing.~P) → AsyncStream[T]

Map an async function to the values in the AsyncStream.

await stream.map(add_one).collect()

async partition(fn: Callable[[P], Awaitable[bool]], *, workers: int = 1, use_threads: bool = False) → tuple[AsyncStream[T], AsyncStream[U]]

Split the AsyncStream into values accepted and rejected by a predicate.

accepted, rejected = await stream.partition(is_valid)

async sequence(*, workers: int = 1, use_threads: bool = False) → Result[Self, E] | Either[Self, E]

Convert a stream of Result or Either values into one monad of stream.

result = await stream.sequence()

tap(fn: ~collections.abc.Callable[[~P], ~collections.abc.Awaitable[None]], *args: ~typing.~P, **kwargs: ~typing.~P) → AsyncStream[T]

Apply an async function to a copy of each value.

The original values remain in the AsyncStream.

await stream.tap(log_value).collect()

class danom.Either(inner: Any = None)

Bases: ABC, Generic

Either monad. Consists of Right and Left for successful and failed operations respectively. Each monad is a frozen instance to prevent further mutation.

abstractmethod and_then(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._either.T_co, ~P]], ~danom._either.Either[~danom._either.U_co, ~danom._either.E_co]], *args: ~typing.~P, **kwargs: ~typing.~P) → Either[U_co, E_co]

Pipe another function that returns a monad. For Left will return original inner.

from danom import Left, Right

Right(1).and_then(add_one) == Right(2)
Right(1).and_then(raise_err) == Left(TypeError())
Left(TypeError()).and_then(add_one) == Left(TypeError())
Left(TypeError()).and_then(raise_value_err) == Left(TypeError())

static either_is_ok(result: Either[T_co, E_co]) → bool

Check whether the monad is ok. Allows for filter or partition in a Stream without needing a lambda or custom function.

from danom import Stream, Either

Stream.from_iterable([Right(), Right(), Left()]).filter(Either.either_is_ok).collect() == (Right(), Right())

static either_unwrap(result: Either[T_co, E_co]) → T_co

Unwrap the Right or Left monad to get the inner value.

from danom import Stream, Either

oks, errs = Stream.from_iterable([Right(1), Right(2), Left()]).partition(Either.either_is_ok)
oks.map(Either.either_unwrap).collect == (1, 2)

flatten() → Either[T_co, E_co]

Flatten the monad. Will return the first Left or the lowest Right instance.

>>> from danom import Left, Right, Stream, Either

>>> Right(Right(Right(1))).flatten() == Right(1)
True

>>> Right(Right(Left())).flatten() == Left()
True

inner: Any

abstractmethod is_ok() → bool

Returns True if the result type is Right. Returns False if the result type is Left.

>>> from danom import Left, Right

>>> Right().is_ok() == True
True

>>> Left().is_ok() == False
True

abstractmethod map(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._either.T_co, ~P]], ~danom._either.U_co], *args: ~typing.~P, **kwargs: ~typing.~P) → Either[U_co, E_co]

Pipe a pure function and wrap the return value with Right. Given an Left will return self.

from danom import Left, Right

Right(1).map(add_one) == Right(2)
Left(1).map(add_one) == Left(1)

abstractmethod map_err(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._either.T_co, ~P]], ~danom._either.U_co], *args: ~typing.~P, **kwargs: ~typing.~P) → Either[U_co, E_co]

Pipe a pure function and wrap the return value with Left. Given an Right will return self.

from danom import Left, Right

Left(TypeError()).map_err(type_err_to_value_err) == Left(ValueError())
Right(1).map(type_err_to_value_err) == Right(1)

abstractmethod or_else(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._either.T_co, ~P]], ~danom._either.Either[~danom._either.U_co, ~danom._either.E_co]], *args: ~typing.~P, **kwargs: ~typing.~P) → Either[U_co, E_co]

Pipe a function that returns a monad to recover from an Left. For Right will return original Either.

from danom import Left, Right

Right(1).or_else(replace_err_with_zero) == Right(1)
Left(TypeError()).or_else(replace_err_with_zero) == Right(0)

classmethod unit(inner: T_co) → Right[T_co]

Unit method. Given an item of type T return Right(T)

>>> from danom import Left, Right, Either

>>> Either.unit(0) == Right(inner=0)
True

>>> Right.unit(0) == Right(inner=0)
True

>>> Left.unit(0) == Right(inner=0)
True

abstractmethod unwrap() → T_co

Unwrap the Right or Left monad to get the inner value.

>>> from danom import Left, Right

>>> Right().unwrap() == None
True

>>> Right(1).unwrap() == 1
True

>>> Right("ok").unwrap() == 'ok'
True

>>> Left(-1).unwrap() == -1
True

class danom.Err(error: Any = None, input_args: tuple[()] | tuple[tuple[Any, ...], dict[str, Any]] | tuple[object, tuple[Any, ...], dict[str, Any]] = (), traceback: str = '')

Bases: Result[Never, E_co]

and_then(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._result.T_co, ~P]], ~danom._result.Result[~danom._result.T_co, ~danom._result.E_co]], *args: ~typing.~P, **kwargs: ~typing.~P) → Self

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

from danom import Err, Ok

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

details: list[dict[str, Any]]

error: Any

input_args: tuple[()] | tuple[tuple[Any, ...], dict[str, Any]] | tuple[object, tuple[Any, ...], dict[str, Any]]

is_ok() → Literal[False]

Returns True if the result type is Ok. Returns False if the result type is Err.

>>> from danom import Err, Ok

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

>>> Err().is_ok() == False
True

map(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._result.T_co, ~P]], ~danom._result.U_co], *args: ~typing.~P, **kwargs: ~typing.~P) → Self

Pipe a pure function and wrap the return value with Ok. Given an Err will return self.

from danom import Err, Ok

Ok(1).map(add_one) == Ok(2)
Err(error=TypeError()).map(add_one) == Err(error=TypeError())

map_err(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._result.T_co, ~P]], ~danom._result.U_co], *args: ~typing.~P, **kwargs: ~typing.~P) → Err[E_co]

Pipe a pure function and wrap the return value with Err. Given an Ok will return self.

from danom import Err, Ok

Err(error=TypeError()).map_err(type_err_to_value_err) == Err(error=ValueError())
Ok(1).map(type_err_to_value_err) == Ok(1)

or_else(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._result.E_co, ~P]], ~danom._result.Result[~danom._result.T_co, ~danom._result.E_co]], *args: ~typing.~P, **kwargs: ~typing.~P) → Result[T_co, E_co]

Pipe a function that returns a monad to recover from an Err. For Ok will return original Result.

from danom import Err, Ok

Ok(1).or_else(replace_err_with_zero) == Ok(1)
Err(error=TypeError()).or_else(replace_err_with_zero) == Ok(0)

traceback: str

unwrap() → T_co

Unwrap the Ok monad and get the inner value. Unwrap the Err monad will raise the inner error.

>>> from danom import Err, Ok

>>> Ok().unwrap() == None
True

>>> Ok(1).unwrap() == 1
True

>>> Ok("ok").unwrap() == 'ok'
True

>>> Err(error=TypeError()).unwrap()
Traceback (most recent call last):
...
TypeError:

class danom.Left(inner: Any = None)

Bases: Either[Never, E_co]

and_then(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._either.T_co, ~P]], ~danom._either.Either[~danom._either.U_co, ~danom._either.E_co]], *args: ~typing.~P, **kwargs: ~typing.~P) → Self

Pipe another function that returns a monad. For Left will return original inner.

from danom import Left, Right

Right(1).and_then(add_one) == Right(2)
Right(1).and_then(raise_err) == Left(TypeError())
Left(TypeError()).and_then(add_one) == Left(TypeError())
Left(TypeError()).and_then(raise_value_err) == Left(TypeError())

is_ok() → Literal[False]

Returns True if the result type is Right. Returns False if the result type is Left.

>>> from danom import Left, Right

>>> Right().is_ok() == True
True

>>> Left().is_ok() == False
True

map(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._either.T_co, ~P]], ~danom._either.U_co], *args: ~typing.~P, **kwargs: ~typing.~P) → Self

Pipe a pure function and wrap the return value with Right. Given an Left will return self.

from danom import Left, Right

Right(1).map(add_one) == Right(2)
Left(1).map(add_one) == Left(1)

map_err(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._either.T_co, ~P]], ~danom._either.U_co], *args: ~typing.~P, **kwargs: ~typing.~P) → Left[F_co]

Pipe a pure function and wrap the return value with Left. Given an Right will return self.

from danom import Left, Right

Left(TypeError()).map_err(type_err_to_value_err) == Left(ValueError())
Right(1).map(type_err_to_value_err) == Right(1)

or_else(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._either.T_co, ~P]], ~danom._either.Either[~danom._either.U_co, ~danom._either.E_co]], *args: ~typing.~P, **kwargs: ~typing.~P) → Either[U_co, E_co]

Pipe a function that returns a monad to recover from an Left. For Right will return original Either.

from danom import Left, Right

Right(1).or_else(replace_err_with_zero) == Right(1)
Left(TypeError()).or_else(replace_err_with_zero) == Right(0)

unwrap() → T_co

Unwrap the Right or Left monad to get the inner value.

>>> from danom import Left, Right

>>> Right().unwrap() == None
True

>>> Right(1).unwrap() == 1
True

>>> Right("ok").unwrap() == 'ok'
True

>>> Left(-1).unwrap() == -1
True

class danom.Ok(inner: Any = None)

Bases: Result[T_co, Never]

and_then(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._result.T_co, ~P]], ~danom._result.Result[~danom._result.T_co, ~danom._result.E_co]], *args: ~typing.~P, **kwargs: ~typing.~P) → Result[T_co, E_co]

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

from danom import Err, Ok

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

inner: Any

is_ok() → Literal[True]

Returns True if the result type is Ok. Returns False if the result type is Err.

>>> from danom import Err, Ok

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

>>> Err().is_ok() == False
True

map(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._result.T_co, ~P]], ~danom._result.U_co], *args: ~typing.~P, **kwargs: ~typing.~P) → Ok[T_co]

Pipe a pure function and wrap the return value with Ok. Given an Err will return self.

from danom import Err, Ok

Ok(1).map(add_one) == Ok(2)
Err(error=TypeError()).map(add_one) == Err(error=TypeError())

map_err(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._result.T_co, ~P]], ~danom._result.U_co], *args: ~typing.~P, **kwargs: ~typing.~P) → Self

Pipe a pure function and wrap the return value with Err. Given an Ok will return self.

from danom import Err, Ok

Err(error=TypeError()).map_err(type_err_to_value_err) == Err(error=ValueError())
Ok(1).map(type_err_to_value_err) == Ok(1)

or_else(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._result.E_co, ~P]], ~danom._result.Result[~danom._result.T_co, ~danom._result.E_co]], *args: ~typing.~P, **kwargs: ~typing.~P) → Self

Pipe a function that returns a monad to recover from an Err. For Ok will return original Result.

from danom import Err, Ok

Ok(1).or_else(replace_err_with_zero) == Ok(1)
Err(error=TypeError()).or_else(replace_err_with_zero) == Ok(0)

unwrap() → T_co

Unwrap the Ok monad and get the inner value. Unwrap the Err monad will raise the inner error.

>>> from danom import Err, Ok

>>> Ok().unwrap() == None
True

>>> Ok(1).unwrap() == 1
True

>>> Ok("ok").unwrap() == 'ok'
True

>>> Err(error=TypeError()).unwrap()
Traceback (most recent call last):
...
TypeError:

class danom.ParStream(seq: tuple, ops: tuple = ())

Bases: _BaseSyncStream, Generic

A stream that applies its operations with a thread or process pool.

Version changes

0.16.0: Added ParStream

collect(*, workers: int = 4, use_threads: bool = False) → tuple[U, ...]

Materialise the ParStream with the configured workers.

workers=-1 uses one worker for each available CPU, except one. Set use_threads to True to use threads instead of processes.

from danom import ParStream

ParStream.from_iterable([1, 2, 3]).map(add_one).collect(workers=2)

to_par() → Self

Return the ParStream unchanged.

to_stream() → Stream[T]

Convert the ParStream to a synchronous Stream.

class danom.Result

Bases: ABC, Generic

Result monad. Consists of Ok and Err for successful and failed operations respectively. Each monad is a frozen instance to prevent further mutation.

abstractmethod and_then(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._result.T_co, ~P]], ~danom._result.Result[~danom._result.T_co, ~danom._result.E_co]], *args: ~typing.~P, **kwargs: ~typing.~P) → Result[T_co, E_co]

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

from danom import Err, Ok

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

flatten() → Result[T_co, E_co]

Flatten the monad. Will return the first Err or the lowest Ok instance.

>>> from danom import Err, Ok, Stream, Result

>>> Ok(Ok(Ok(1))).flatten() == Ok(1)
True

>>> Ok(Ok(Err())).flatten() == Err()
True

abstractmethod is_ok() → bool

Returns True if the result type is Ok. Returns False if the result type is Err.

>>> from danom import Err, Ok

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

>>> Err().is_ok() == False
True

abstractmethod map(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._result.T_co, ~P]], ~danom._result.U_co], *args: ~typing.~P, **kwargs: ~typing.~P) → Result[T_co, E_co]

Pipe a pure function and wrap the return value with Ok. Given an Err will return self.

from danom import Err, Ok

Ok(1).map(add_one) == Ok(2)
Err(error=TypeError()).map(add_one) == Err(error=TypeError())

abstractmethod map_err(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._result.T_co, ~P]], ~danom._result.U_co], *args: ~typing.~P, **kwargs: ~typing.~P) → Result[T_co, E_co]

Pipe a pure function and wrap the return value with Err. Given an Ok will return self.

from danom import Err, Ok

Err(error=TypeError()).map_err(type_err_to_value_err) == Err(error=ValueError())
Ok(1).map(type_err_to_value_err) == Ok(1)

abstractmethod or_else(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._result.E_co, ~P]], ~danom._result.Result[~danom._result.T_co, ~danom._result.E_co]], *args: ~typing.~P, **kwargs: ~typing.~P) → Result[T_co, E_co]

Pipe a function that returns a monad to recover from an Err. For Ok will return original Result.

from danom import Err, Ok

Ok(1).or_else(replace_err_with_zero) == Ok(1)
Err(error=TypeError()).or_else(replace_err_with_zero) == Ok(0)

static result_is_ok(result: Result[T_co, E_co]) → bool

Check whether the monad is ok. Allows for filter or partition in a Stream without needing a lambda or custom function.

from danom import Stream, Result

Stream.from_iterable([Ok(), Ok(), Err()]).filter(Result.result_is_ok).collect() == (Ok(), Ok())

static result_unwrap(result: Result[T_co, E_co]) → T_co

Unwrap the Ok monad and get the inner value. Unwrap the Err monad will raise the inner error.

from danom import Err, Ok, Stream, Result

oks, errs = Stream.from_iterable([Ok(1), Ok(2), Err()]).partition(Result.result_is_ok)
oks.map(Result.result_unwrap).collect == (1, 2)

classmethod unit(inner: T_co) → Ok[T_co]

Unit method. Given an item of type T return Ok(T)

>>> from danom import Err, Ok, Result

>>> Result.unit(0) == Ok(0)
True

>>> Ok.unit(0) == Ok(0)
True

>>> Err.unit(0) == Ok(0)
True

abstractmethod unwrap() → T_co

Unwrap the Ok monad and get the inner value. Unwrap the Err monad will raise the inner error.

>>> from danom import Err, Ok

>>> Ok().unwrap() == None
True

>>> Ok(1).unwrap() == 1
True

>>> Ok("ok").unwrap() == 'ok'
True

>>> Err(error=TypeError()).unwrap()
Traceback (most recent call last):
...
TypeError:

class danom.Right(inner: Any = None)

Bases: Either[T_co, Never]

and_then(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._either.T_co, ~P]], ~danom._either.Either[~danom._either.U_co, ~danom._either.E_co]], *args: ~typing.~P, **kwargs: ~typing.~P) → Either[U_co, E_co]

Pipe another function that returns a monad. For Left will return original inner.

from danom import Left, Right

Right(1).and_then(add_one) == Right(2)
Right(1).and_then(raise_err) == Left(TypeError())
Left(TypeError()).and_then(add_one) == Left(TypeError())
Left(TypeError()).and_then(raise_value_err) == Left(TypeError())

is_ok() → Literal[True]

Returns True if the result type is Right. Returns False if the result type is Left.

>>> from danom import Left, Right

>>> Right().is_ok() == True
True

>>> Left().is_ok() == False
True

map(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._either.T_co, ~P]], ~danom._either.U_co], *args: ~typing.~P, **kwargs: ~typing.~P) → Right[U_co]

Pipe a pure function and wrap the return value with Right. Given an Left will return self.

from danom import Left, Right

Right(1).map(add_one) == Right(2)
Left(1).map(add_one) == Left(1)

map_err(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._either.T_co, ~P]], ~danom._either.U_co], *args: ~typing.~P, **kwargs: ~typing.~P) → Self

Pipe a pure function and wrap the return value with Left. Given an Right will return self.

from danom import Left, Right

Left(TypeError()).map_err(type_err_to_value_err) == Left(ValueError())
Right(1).map(type_err_to_value_err) == Right(1)

or_else(func: ~collections.abc.Callable[[~typing.Concatenate[~danom._either.T_co, ~P]], ~danom._either.Either[~danom._either.U_co, ~danom._either.E_co]], *args: ~typing.~P, **kwargs: ~typing.~P) → Self

Pipe a function that returns a monad to recover from an Left. For Right will return original Either.

from danom import Left, Right

Right(1).or_else(replace_err_with_zero) == Right(1)
Left(TypeError()).or_else(replace_err_with_zero) == Right(0)

unwrap() → T_co

Unwrap the Right or Left monad to get the inner value.

>>> from danom import Left, Right

>>> Right().unwrap() == None
True

>>> Right(1).unwrap() == 1
True

>>> Right("ok").unwrap() == 'ok'
True

>>> Left(-1).unwrap() == -1
True

class danom.Stream(seq: tuple, ops: tuple = ())

Bases: _BaseSyncStream, Generic

A lazy iterator with functional operations.

Why bother?

Readability counts, abstracting common operations helps reduce cognitive complexity when reading code.

Comparison

Take this imperative pipeline of operations, it iterates once over the data, skipping the value if it fails one of the filter checks:

res = []

for x in range(1_000_000):
    item = triple(x)

    if not is_gt_ten(item):
        continue

    item = min_two(item)

    if not is_even_num(item):
        continue

    item = square(item)

    if not is_lt_400(item):
        continue

    res.append(item)
[100, 256]

number of tokens: 90

number of keywords: 11

keyword breakdown: {‘for’: 1, ‘in’: 1, ‘if’: 3, ‘not’: 3, ‘continue’: 3}

After a bit of experience with python you might use list comprehensions, however this is arguably _less_ clear and iterates multiple times over the same data

mul_three = [triple(x) for x in range(1_000_000)]
gt_ten = [x for x in mul_three if is_gt_ten(x)]
sub_two = [min_two(x) for x in gt_ten]
is_even = [x for x in sub_two if is_even_num(x)]
squared = [square(x) for x in is_even]
lt_400 = [x for x in squared if is_lt_400(x)]
[100, 256]

number of tokens: 92

number of keywords: 15

keyword breakdown: {‘for’: 6, ‘in’: 6, ‘if’: 3}

This still has a lot of tokens that the developer has to read to understand the code. The extra keywords add noise that cloud the actual transformations.

Using a Stream results in this:

from danom import Stream

(
    Stream.from_iterable(range(1_000_000))
    .map(triple)
    .filter(is_gt_ten)
    .map(min_two)
    .filter(is_even_num)
    .map(square)
    .filter(is_lt_400)
    .collect()
)
(100, 256)

number of tokens: 60

number of keywords: 0

keyword breakdown: {}

The business logic is arguably much clearer like this.

Version changes

0.13.0: Stream.map, Stream.filter and Stream.tap now take kwargs and partial them into the passed in function.

collect(*, workers: int = 4, use_threads: bool = False) → tuple[U, ...]

Materialise the sequence from the Stream.

from danom import Stream

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

to_par() → ParStream[T]

Convert the Stream to a ParStream.

to_stream() → Stream[T]

Return the Stream unchanged.

danom.all_of(*fns: Callable[[T_co], bool]) → Callable[[T_co], bool]

True if all of the given functions return True.

from danom import all_of

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

danom.any_of(*fns: Callable[[T_co], bool]) → Callable[[T_co], bool]

True if any of the given functions return True.

from danom import any_of

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

danom.compose(*fns: Callable[[T_co], T_co | U_co]) → Callable[[T_co], T_co | U_co]

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.

from danom import compose

add_two = compose(add_one, add_one)
add_two(0) == 2
add_two_is_even = compose(add_one, add_one, is_even)
add_two_is_even(0) == True

danom.identity(x: T_co) → T_co

Basic identity function.

from danom import identity

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

Papertrail examples:

>>> identity(1) == 1
True

>>> identity("abc") == "abc"
True

>>> identity([0, 1, 2]) == [0, 1, 2]
True

>>> identity(Ok(inner=1)) == Ok(inner=1)
True

danom.invert(func: Callable[[T_co], bool]) → Callable[[T_co], bool]

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

from danom import invert

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

danom.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.

>>> from danom import new_type

>>> def is_positive[T](value: T) -> bool:
...     return value >= 0

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

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.

from danom import new_type

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

Email = new_type("Email", str, validators=[has_len])
Email("some_email@domain.com").upper() == "SOME_EMAIL@DOMAIN.COM"
Email("some_email@domain.com").map(str.upper) == Email(inner='SOME_EMAIL@DOMAIN.COM')

danom.none_of(*fns: Callable[[T_co], bool]) → Callable[[T_co], bool]

True if none of the given functions return True.

from danom import none_of

is_valid = none_of(is_empty, exceeds_size_limit, contains_unsupported_format)
is_valid(submission) == True

danom.safe(func: Callable[[P], U]) → Callable[[P], Result[U, Exception]]

danom.safe(func: None = None, *, errors: ExceptionType = Exception) → Callable[[Callable[[P], U]], Callable[[P], Result[U, Exception]]]

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

from danom import safe

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

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

Only catch a single error type or subset of error types by passing in an error type to catch.

from danom import safe

@safe(errors=ZeroDivisionError)
def div(a: int, b: int) -> float:
    return a / b


div(2, 0) == Err(error=ZeroDivisionError('division by zero'))
div(2, "")
---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)

danom.safe_method(func: Callable[[Concatenate[T, P]], U]) → Callable[[Concatenate[T, P]], Result[U, Exception]]

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

from danom import safe_method

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)

Indices and tables

• Index

• Module Index

• Search Page