Library of functions for iterating iterables.¶

Concatenating and merging iterables
Dropping and taking values from iterables
Reducing and accumulating iterables
Assumptions
- iterables are not necessarily iterators
- at all times iterator protocol is assumed to be followed
  - all iterators are assumed to be iterable
  - for all iterators foo we assume iter(foo) is foo

class pythonic_fp.iterables.FM(*values)¶

Iterable Blending Enums.

CONCAT: Concatenate iterables first to last
MERGE: Merge iterables until one is exhausted
EXHAUST: Merge iterables until all are exhausted

pythonic_fp.iterables.accumulate(iterable, f, initial=NoValue())¶

Returns an iterator of partial fold values.

def accumulate[D, L](
    iterable: Iterable[D],
    f: Callable[[L, D], L],
    initial: L
) -> Iterator[L]

A pure Python version of standard library’s itertools.accumulate

function f does not default to addition (for typing flexibility)
begins accumulation with an “optional” initial value

Parameters:

iterable (Iterable[TypeVar(D)]) – iterable to be folded
f (Callable[[TypeVar(L), TypeVar(D)], TypeVar(L)]) – 2 parameter function, first parameter for accumulated value
initial (TypeVar(L) | NoValue) – “optional” initial value to start fold

Return type:

Iterator[TypeVar(L)]

Returns:

an iterator of the intermediate fold values

pythonic_fp.iterables.concat(*iterables)¶

Sequentially concatenate multiple iterables together.

def concat[D](iterable: Iterable[D]) -> Iterator[D]

pure Python version of standard library’s itertools.chain
iterator sequentially yields each iterable until all are exhausted
an infinite iterable will prevent subsequent iterables from yielding any values
performant to itertools.chain

Parameters:: iterables (Iterable[TypeVar(D)]) – iterables to concatenate
Return type:: Iterator[TypeVar(D)]
Returns:: concatenated iterators

pythonic_fp.iterables.drop(iterable, n)¶

Drop the next n values from iterable.

def drop[D](
    iterable: Iterable[D],
    n: int
) -> Iterator[D]

Parameters:

iterable (Iterable[TypeVar(D)]) – iterable whose values are to be dropped
n (int) – number of values to be dropped

Return type:

Iterator[TypeVar(D)]

Returns:

an iterator of the remaining values

pythonic_fp.iterables.drop_while(iterable, pred)¶

Drop initial values from iterable while predicate is true.

def drop_while[D](
    iterable: Iterable[D],
    pred: Callable[[D], bool]
) -> Iterator[D]

Parameters:

iterable (Iterable[TypeVar(D)]) – iterable whose values are to be dropped
pred (Callable[[TypeVar(D)], bool]) – Boolean valued function

Return type:

Iterator[TypeVar(D)]

Returns:

an iterator beginning where pred returned false

pythonic_fp.iterables.exhaust(*iterables)¶

Shuffle together multiple iterables until all are exhausted.

def exhaust[D](iterable: Iterable[D]) -> Iterator[D]

Parameters:: iterables (Iterable[TypeVar(D)]) – iterables to exhaustively merge
Return type:: Iterator[TypeVar(D)]
Returns:: merged iterators

pythonic_fp.iterables.fold_left(iterable, f, initial)¶

Fold an iterable left with a function and initial value.

def fold_left[D, L](
    iterable: Iterable[D],
    f: Callable[[L, D], L],
    initial: L
) -> L | Never

not restricted to __add__ for the folding function
initial value required, does not default to 0 for initial value
handles non-numeric data just find

Warning

This function never return if given an infinite iterable.

Warning

This function does not catch any exceptions f may raise.

Parameters:

iterable (Iterable[TypeVar(D)]) – iterable to be folded
f (Callable[[TypeVar(L), TypeVar(D)], TypeVar(L)]) – 2 parameter function, first parameter for accumulated value
initial (TypeVar(L)) – mandatory initial value to start fold

Return type:

TypeVar(L) | Never

Returns:

the folded value

pythonic_fp.iterables.maybe_fold_left(iterable, f, initial=NoValue())¶

Folds an iterable left with an “optional” initial value..

def maybe_fold_left[D, L](
    iterable: Iterable[D],
    f: Callable[[L, D], L],
    initial: L
) -> MayBe[L] | Never

traditional FP type order given for function f
when an initial value is not given then L = D
if iterable empty and no initial value given, return MayBe()

Warning

This function never return if given an infinite iterable.

Warning

This function does not catch any exceptions f may raise.

Parameters:

iterable (Iterable[TypeVar(D)]) – The iterable to be folded.
f (Callable[[TypeVar(L), TypeVar(D)], TypeVar(L)]) – First argument is for the accumulated value.
initial (TypeVar(L) | NoValue) – Mandatory initial value to start fold.

Return type:

MayBe[TypeVar(L)] | Never

Returns:

Folded value if it exists.

pythonic_fp.iterables.merge(*iterables, yield_partials=False)¶

Shuffle together iterables until one is exhausted.

def merge[D](iterable: Iterable[D], yield_partials: bool) -> Iterator[D]

If yield_partials is true

yield any unmatched yielded values from other iterables
prevents data lose if any of the iterables are iterators with external references

Parameters:

iterables (Iterable[TypeVar(D)]) – iterables to merge until one gets exhausted
yield_partials (bool)

Return type:

Iterator[TypeVar(D)]

Returns:

merged iterators

pythonic_fp.iterables.reduce_left(iterable, f)¶

Fold an iterable left with a function.

def reduce_left[D](
    iterable: Iterable[D],
    f: Callable[[D, D], D]
) -> D | Never

Warning

This function never return if given an infinite iterable.

Warning

This function does not catch or re-raises exceptions from f.

Parameters:

iterable (Iterable[TypeVar(D)]) – iterable to be reduced (folded)
f (Callable[[TypeVar(D), TypeVar(D)], TypeVar(D)]) – 2 parameter function, first parameter for accumulated value

Return type:

TypeVar(D) | Never

Returns:

reduced value from the iterable

Raises:

StopIteration – when called on an empty circular array
Exception – does not catch or re-raises exceptions from f.

pythonic_fp.iterables.sc_reduce_left(iterable, f, start=<function <lambda>>, stop=<function <lambda>>, include_start=True, include_stop=True)¶

Short circuit version of a left reduce.

def sc_reduce_left[D](
    iterable: Iterable[D],
    f: Callable[[D, D], D],
    start: Callable[[D], bool],
    stop: Callable[[D], bool],
    include_start: bool,
    include_stop: bool
) -> tuple[MayBe[D], Iterator[D]]

Useful for infinite iterables.

Behavior for default arguments will

left reduce finite iterable
start folding immediately
continue folding until end (of a possibly infinite iterable)

Parameters:

iterable (Iterable[TypeVar(D)]) – iterable to be reduced from the left
f (Callable[[TypeVar(D), TypeVar(D)], TypeVar(D)]) – 2 parameter function, first parameter for accumulated value
start (Callable[[TypeVar(D)], bool]) – delay starting the fold until it returns true
stop (Callable[[TypeVar(D)], bool]) – prematurely stop the fold when it returns true
include_start (bool) – if true, include fold starting value in fold
include_stop (bool) – if true, include stopping value in fold

Return type:

tuple[MayBe[TypeVar(D)], Iterator[TypeVar(D)]]

Returns:

MayBe of the folded value and remaining iterables

pythonic_fp.iterables.sc_reduce_right(iterable, f, start=<function <lambda>>, stop=<function <lambda>>, include_start=True, include_stop=True)¶

Short circuit version of a right reduce.

def sc_reduce_left[D](
    iterable: Iterable[D],
    f: Callable[[D, D], D],
    start: Callable[[D], bool],
    stop: Callable[[D], bool],
    include_start: bool,
    include_stop: bool
) -> Tuple[MayBe[D] | Iterator[D]

Useful for infinite and non-reversible iterables.

Behavior for default arguments will

right reduce finite iterable
start folding at end (of a possibly infinite iterable)
continue reducing right until beginning

Parameters:

iterable (Iterable[TypeVar(D)]) – iterable to be reduced from the right
f (Callable[[TypeVar(D), TypeVar(D)], TypeVar(D)]) – 2 parameter function, second parameter for accumulated value
start (Callable[[TypeVar(D)], bool]) – delay starting the fold until it returns true
stop (Callable[[TypeVar(D)], bool]) – prematurely stop the fold when it returns true
include_start (bool) – if true, include fold starting value in fold
include_stop (bool) – if true, include stopping value in fold

Return type:

tuple[MayBe[TypeVar(D)], Iterator[TypeVar(D)]]

Returns:

MayBe of the folded value and remaining iterables

pythonic_fp.iterables.take(iterable, n)¶

Return an iterator of up to n initial values of an iterable.

def take[D](
    iterable: Iterable[D],
    n: int
) -> Iterator[D]

Parameters:

iterable (Iterable[TypeVar(D)]) – iterable providing the values to be taken
n (int) – number of values to be dropped

Return type:

Iterator[TypeVar(D)]

Returns:

an iterator of up to n initial values from iterable

pythonic_fp.iterables.take_split(iterable, n)¶

Same as take except also return iterator of remaining values.

def take_split[D](
    iterable: Iterable[D],
    n: int
) -> tuple[Iterator[D], Iterator[D]]

Warning

CONTRACT: Do not access the second returned iterator until the first one is exhausted.

Parameters:

iterable (Iterable[TypeVar(D)]) – iterable providing the values to be taken
n (int) – maximum number of values to be taken

Return type:

tuple[Iterator[TypeVar(D)], Iterator[TypeVar(D)]]

Returns:

an iterator of values taken and an iterator of remaining values

pythonic_fp.iterables.take_while(iterable, pred)¶

Yield values from iterable while predicate is true.

def take_while[D](
    iterable: Iterable[D],
    pred: Callable[[D], bool]
) -> Iterator[D]

Warning

Risk of value loss if iterable is multiple referenced iterator.

Parameters:

iterable (Iterable[TypeVar(D)]) – iterable providing the values to be taken
pred (Callable[[TypeVar(D)], bool]) – Boolean valued function

Return type:

Iterator[TypeVar(D)]

Returns:

an Iterator of up to n initial values from the iterable

pythonic_fp.iterables.take_while_split(iterable, pred)¶

Yield values from iterable while predicate is true.

def take_while_split[D](
    iterable: Iterable[D],
    pred: Callable[[D], bool]
) -> tuple[Iterator[D], Iterator[D]]

Warning

CONTRACT: Do not access the second returned iterator until the first one is exhausted.

Parameters:

iterable (Iterable[TypeVar(D)]) – iterable providing the values to be taken
pred (Callable[[TypeVar(D)], bool]) – Boolean valued function

Return type:

tuple[Iterator[TypeVar(D)], Iterator[TypeVar(D)]]

Returns:

an iterator of values taken and an iterator of remaining values