Seq¶

Bases: BaseAgg[T], BaseEager[T]

pyochain.Seq represent an in memory collection.

Provides a subset of pyochain.Iter methods with eager evaluation, and is the return type of pyochain.Iter.collect().

Source code in src/pyochain/_iter/_main.py

class Seq[T](BaseAgg[T], BaseEager[T]):
    """
    pyochain.Seq represent an in memory collection.

    Provides a subset of pyochain.Iter methods with eager evaluation, and is the return type of pyochain.Iter.collect().
    """

    __slots__ = ("_data",)

    def __init__(self, data: Collection[T]) -> None:
        self._data = data

    def iter(self) -> Iter[T]:
        """
        Get an iterator over the sequence.
        Call this to switch to lazy evaluation.
        """
        return Iter.from_(self.unwrap())

slots `class-attribute` `instance-attribute` ¶

__slots__ = ('_data',)

_data `instance-attribute` ¶

_data = data

init ¶

__init__(data: Collection[T]) -> None

Source code in src/pyochain/_iter/_main.py

def __init__(self, data: Collection[T]) -> None:
    self._data = data

apply ¶

apply(
    func: Callable[Concatenate[Iterable[T], P], Iterator[R]],
    *args: P.args,
    **kwargs: P.kwargs,
) -> Iter[R]

Apply a function to the underlying iterable and return an Iter of the result. Allow to pass user defined functions that transform the iterable while retaining the Iter wrapper.

Parameters:

Name	Type	Description	Default
`func`	`Callable[Concatenate[Iterable[T], P], Iterator[R]]`	Function to apply to the underlying iterable.	required
`*args`	`P.args`	Positional arguments to pass to the function.	`()`
`**kwargs`	`P.kwargs`	Keyword arguments to pass to the function.	`{}`

Example:

>>> import pyochain as pc
>>> def double(data: Iterable[int]) -> Iterator[int]:
...     return (x * 2 for x in data)
>>> pc.Iter.from_([1, 2, 3]).apply(double).into(list)
[2, 4, 6]

Source code in src/pyochain/_core/_main.py

def apply[**P, R](
    self,
    func: Callable[Concatenate[Iterable[T], P], Iterator[R]],
    *args: P.args,
    **kwargs: P.kwargs,
) -> Iter[R]:
    """
    Apply a function to the underlying iterable and return an Iter of the result.
    Allow to pass user defined functions that transform the iterable while retaining the Iter wrapper.

    Args:
        func: Function to apply to the underlying iterable.
        *args: Positional arguments to pass to the function.
        **kwargs: Keyword arguments to pass to the function.

    Example:
    ```python
    >>> import pyochain as pc
    >>> def double(data: Iterable[int]) -> Iterator[int]:
    ...     return (x * 2 for x in data)
    >>> pc.Iter.from_([1, 2, 3]).apply(double).into(list)
    [2, 4, 6]

    ```
    """
    from .._iter import Iter

    return Iter(self.into(func, *args, **kwargs))

argmax ¶

argmax(key: Callable[[T], U] | None = None) -> int

Index of the first occurrence of a maximum value in an iterable.

Parameters:

Name	Type	Description	Default
`key`	`Callable[[T], U] \| None`	Optional function to determine the value for comparison.	`None`

>>> import pyochain as pc
>>> pc.Iter.from_("abcdefghabcd").argmax()
7
>>> pc.Iter.from_([0, 1, 2, 3, 3, 2, 1, 0]).argmax()
3

For example, identify the best machine learning model:

>>> models = pc.Iter.from_(["svm", "random forest", "knn", "naïve bayes"])
>>> accuracy = pc.Seq([68, 61, 84, 72])
>>> # Most accurate model
>>> models.item(accuracy.argmax())
'knn'
>>>
>>> # Best accuracy
>>> accuracy.into(max)
84

Source code in src/pyochain/_iter/_aggregations.py

def argmax[U](self, key: Callable[[T], U] | None = None) -> int:
    """
    Index of the first occurrence of a maximum value in an iterable.

    Args:
        key: Optional function to determine the value for comparison.

    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_("abcdefghabcd").argmax()
    7
    >>> pc.Iter.from_([0, 1, 2, 3, 3, 2, 1, 0]).argmax()
    3

    ```
    For example, identify the best machine learning model:
    ```python
    >>> models = pc.Iter.from_(["svm", "random forest", "knn", "naïve bayes"])
    >>> accuracy = pc.Seq([68, 61, 84, 72])
    >>> # Most accurate model
    >>> models.item(accuracy.argmax())
    'knn'
    >>>
    >>> # Best accuracy
    >>> accuracy.into(max)
    84

    ```
    """
    return self.into(mit.argmax, key=key)

argmin ¶

argmin(key: Callable[[T], U] | None = None) -> int

Index of the first occurrence of a minimum value in an iterable.

Parameters:

Name	Type	Description	Default
`key`	`Callable[[T], U] \| None`	Optional function to determine the value for comparison.	`None`

>>> import pyochain as pc
>>> pc.Iter.from_("efghabcdijkl").argmin()
4
>>> pc.Iter.from_([3, 2, 1, 0, 4, 2, 1, 0]).argmin()
3

For example, look up a label corresponding to the position of a value that minimizes a cost function:

>>> def cost(x):
...     "Days for a wound to heal given a subject's age."
...     return x**2 - 20 * x + 150
>>> labels = pc.Iter.from_(["homer", "marge", "bart", "lisa", "maggie"])
>>> ages = pc.Seq([35, 30, 10, 9, 1])
>>> # Fastest healing family member
>>> labels.item(ages.argmin(key=cost))
'bart'
>>> # Age with fastest healing
>>> ages.into(min, key=cost)
10

Source code in src/pyochain/_iter/_aggregations.py

def argmin[U](self, key: Callable[[T], U] | None = None) -> int:
    """
    Index of the first occurrence of a minimum value in an iterable.

    Args:
        key: Optional function to determine the value for comparison.

    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_("efghabcdijkl").argmin()
    4
    >>> pc.Iter.from_([3, 2, 1, 0, 4, 2, 1, 0]).argmin()
    3

    ```

    For example, look up a label corresponding to the position of a value that minimizes a cost function:
    ```python
    >>> def cost(x):
    ...     "Days for a wound to heal given a subject's age."
    ...     return x**2 - 20 * x + 150
    >>> labels = pc.Iter.from_(["homer", "marge", "bart", "lisa", "maggie"])
    >>> ages = pc.Seq([35, 30, 10, 9, 1])
    >>> # Fastest healing family member
    >>> labels.item(ages.argmin(key=cost))
    'bart'
    >>> # Age with fastest healing
    >>> ages.into(min, key=cost)
    10

    ```
    """
    return self.into(mit.argmin, key=key)

collect ¶

collect(factory: Callable[[Iterable[T]], Collection[T]] = list) -> Seq[T]

Collect the elements into a sequence.

Parameters:

Name	Type	Description	Default
`factory`	`Callable[[Iterable[T]], Collection[T]]`	A callable that takes an iterable and returns a collection. Defaults to list.	`list`

Example:

>>> import pyochain as pc
>>> pc.Iter.from_(range(5)).collect().unwrap()
[0, 1, 2, 3, 4]

Source code in src/pyochain/_core/_main.py

def collect(self, factory: Callable[[Iterable[T]], Collection[T]] = list) -> Seq[T]:
    """
    Collect the elements into a sequence.

    Args:
        factory: A callable that takes an iterable and returns a collection. Defaults to list.

    Example:
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_(range(5)).collect().unwrap()
    [0, 1, 2, 3, 4]

    ```
    """
    from .._iter import Seq

    return Seq(self.into(factory))

combination_index ¶

combination_index(r: Iterable[T]) -> int

Equivalent to list(combinations(iterable, r)).index(element).

Parameters:

Name	Type	Description	Default
`r`	`Iterable[T]`	The combination to find the index of.	required

The subsequences of iterable that are of length r can be ordered lexicographically.

combination_index computes the index of the first element, without computing the previous combinations.

ValueError will be raised if the given element isn't one of the combinations of iterable.

>>> import pyochain as pc
>>> pc.Iter.from_("abcdefg").combination_index("adf")
10

Source code in src/pyochain/_iter/_aggregations.py

def combination_index(self, r: Iterable[T]) -> int:
    """
    Equivalent to list(combinations(iterable, r)).index(element).

    Args:
        r: The combination to find the index of.

    The subsequences of iterable that are of length r can be ordered lexicographically.

    combination_index computes the index of the first element, without computing the previous combinations.

    ValueError will be raised if the given element isn't one of the combinations of iterable.
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_("abcdefg").combination_index("adf")
    10

    ```
    """
    return self.into(functools.partial(mit.combination_index, r))

count ¶

count() -> int

Return the length of the sequence. Like the builtin len but works on lazy sequences.

>>> import pyochain as pc
>>> pc.Iter.from_([1, 2]).count()
2

Source code in src/pyochain/_iter/_aggregations.py

def count(self) -> int:
    """
    Return the length of the sequence.
    Like the builtin len but works on lazy sequences.
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([1, 2]).count()
    2

    ```
    """
    return self.into(cz.itertoolz.count)

diff_symmetric ¶

diff_symmetric(*others: Iterable[T]) -> Seq[T]

Return the symmetric difference (XOR) of this iterable and 'others'.

Note

This method consumes inner data, unsorts it, and removes duplicates.

Parameters:

Name	Type	Description	Default
`*others`	`Iterable[T]`	Other iterables to compute the symmetric difference with.	`()`

Example:

>>> import pyochain as pc
>>> pc.Iter.from_([1, 2, 2]).diff_symmetric([2, 3]).iter().sort().unwrap()
[1, 3]
>>> pc.Iter.from_([1, 2, 3]).diff_symmetric([3, 4, 5]).iter().sort().unwrap()
[1, 2, 4, 5]

Source code in src/pyochain/_iter/_eager.py

def diff_symmetric(self, *others: Iterable[T]) -> Seq[T]:
    """
    Return the symmetric difference (XOR) of this iterable and 'others'.

    Note:
        This method consumes inner data, unsorts it, and removes duplicates.

    Args:
        *others: Other iterables to compute the symmetric difference with.
    Example:
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([1, 2, 2]).diff_symmetric([2, 3]).iter().sort().unwrap()
    [1, 3]
    >>> pc.Iter.from_([1, 2, 3]).diff_symmetric([3, 4, 5]).iter().sort().unwrap()
    [1, 2, 4, 5]

    ```
    """

    def _symmetric_difference(data: Iterable[T]) -> set[T]:
        return set(data).symmetric_difference(*others)

    return self.collect(_symmetric_difference)

diff_unique ¶

diff_unique(*others: Iterable[T]) -> Seq[T]

Return the difference of this iterable and 'others'. (Elements in 'self' but not in 'others').

Note

This method consumes inner data, unsorts it, and removes duplicates.

Parameters:

Name	Type	Description	Default
`*others`	`Iterable[T]`	Other iterables to subtract from this iterable.	`()`

Example:

>>> import pyochain as pc
>>> pc.Iter.from_([1, 2, 2]).diff_unique([2, 3]).unwrap()
{1}

Source code in src/pyochain/_iter/_eager.py

def diff_unique(self, *others: Iterable[T]) -> Seq[T]:
    """
    Return the difference of this iterable and 'others'.
    (Elements in 'self' but not in 'others').

    Note:
        This method consumes inner data, unsorts it, and removes duplicates.

    Args:
        *others: Other iterables to subtract from this iterable.
    Example:
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([1, 2, 2]).diff_unique([2, 3]).unwrap()
    {1}

    ```
    """

    def _difference(data: Iterable[T]) -> set[T]:
        return set(data).difference(*others)

    return self.collect(_difference)

first ¶

first() -> T

Return the first element.

>>> import pyochain as pc
>>> pc.Iter.from_([9]).first()
9

Source code in src/pyochain/_iter/_aggregations.py

def first(self) -> T:
    """
    Return the first element.
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([9]).first()
    9

    ```
    """
    return self.into(cz.itertoolz.first)

intersection ¶

intersection(*others: Iterable[T]) -> Seq[T]

Return the elements common to this iterable and 'others'.

Note

This method consumes inner data, unsorts it, and removes duplicates.

Parameters:

Name	Type	Description	Default
`*others`	`Iterable[T]`	Other iterables to intersect with.	`()`

Example:

>>> import pyochain as pc
>>> pc.Iter.from_([1, 2, 2]).intersection([2, 3], [2]).unwrap()
{2}

Source code in src/pyochain/_iter/_eager.py

def intersection(self, *others: Iterable[T]) -> Seq[T]:
    """
    Return the elements common to this iterable and 'others'.

    Note:
        This method consumes inner data, unsorts it, and removes duplicates.

    Args:
        *others: Other iterables to intersect with.
    Example:
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([1, 2, 2]).intersection([2, 3], [2]).unwrap()
    {2}

    ```
    """

    def _intersection(data: Iterable[T]) -> set[T]:
        return set(data).intersection(*others)

    return self.collect(_intersection)

into ¶

into(func: Callable[Concatenate[T, P], R], *args: P.args, **kwargs: P.kwargs) -> R

Pass the unwrapped underlying data into a function.

The result is not wrapped.

>>> import pyochain as pc
>>> pc.Iter.from_(range(5)).into(list)
[0, 1, 2, 3, 4]

This is a core functionality that allows ending the chain whilst keeping the code style consistent.

Source code in src/pyochain/_core/_main.py

def into[**P, R](
    self,
    func: Callable[Concatenate[T, P], R],
    *args: P.args,
    **kwargs: P.kwargs,
) -> R:
    """
    Pass the *unwrapped* underlying data into a function.

    The result is not wrapped.
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_(range(5)).into(list)
    [0, 1, 2, 3, 4]

    ```
    This is a core functionality that allows ending the chain whilst keeping the code style consistent.
    """
    return func(self.unwrap(), *args, **kwargs)

item ¶

item(index: int) -> T

Return item at index.

Parameters:

Name	Type	Description	Default
`index`	`int`	The index of the item to retrieve.	required

>>> import pyochain as pc
>>> pc.Iter.from_([10, 20]).item(1)
20

Source code in src/pyochain/_iter/_aggregations.py

def item(self, index: int) -> T:
    """
    Return item at index.

    Args:
        index: The index of the item to retrieve.

    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([10, 20]).item(1)
    20

    ```
    """
    return self.into(functools.partial(cz.itertoolz.nth, index))

iter ¶

iter() -> Iter[T]

Get an iterator over the sequence. Call this to switch to lazy evaluation.

Source code in src/pyochain/_iter/_main.py

def iter(self) -> Iter[T]:
    """
    Get an iterator over the sequence.
    Call this to switch to lazy evaluation.
    """
    return Iter.from_(self.unwrap())

last ¶

last() -> T

Return the last element.

>>> import pyochain as pc
>>> pc.Iter.from_([7, 8, 9]).last()
9

Source code in src/pyochain/_iter/_aggregations.py

def last(self) -> T:
    """
    Return the last element.
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([7, 8, 9]).last()
    9

    ```
    """
    return self.into(cz.itertoolz.last)

max ¶

max() -> U

Return the maximum of the sequence.

>>> import pyochain as pc
>>> pc.Iter.from_([3, 1, 2]).max()
3

Source code in src/pyochain/_iter/_aggregations.py

def max[U: int | float](self: IterWrapper[U]) -> U:
    """
    Return the maximum of the sequence.
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([3, 1, 2]).max()
    3

    ```
    """
    return self.into(max)

mean ¶

mean() -> float

Return the mean of the sequence.

>>> import pyochain as pc
>>> pc.Iter.from_([1, 2, 3]).mean()
2

Source code in src/pyochain/_iter/_aggregations.py

def mean[U: int | float](self: IterWrapper[U]) -> float:
    """
    Return the mean of the sequence.
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([1, 2, 3]).mean()
    2

    ```
    """
    return self.into(statistics.mean)

median ¶

median() -> float

Return the median of the sequence.

>>> import pyochain as pc
>>> pc.Iter.from_([1, 3, 2]).median()
2

Source code in src/pyochain/_iter/_aggregations.py

def median[U: int | float](self: IterWrapper[U]) -> float:
    """
    Return the median of the sequence.
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([1, 3, 2]).median()
    2

    ```
    """
    return self.into(statistics.median)

min ¶

min() -> U

Return the minimum of the sequence.

>>> import pyochain as pc
>>> pc.Iter.from_([3, 1, 2]).min()
1

Source code in src/pyochain/_iter/_aggregations.py

def min[U: int | float](self: IterWrapper[U]) -> U:
    """
    Return the minimum of the sequence.
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([3, 1, 2]).min()
    1

    ```
    """
    return self.into(min)

mode ¶

mode() -> U

Return the mode of the sequence.

>>> import pyochain as pc
>>> pc.Iter.from_([1, 2, 2, 3]).mode()
2

Source code in src/pyochain/_iter/_aggregations.py

def mode[U: int | float](self: IterWrapper[U]) -> U:
    """
    Return the mode of the sequence.
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([1, 2, 2, 3]).mode()
    2

    ```
    """
    return self.into(statistics.mode)

most_common ¶

most_common(n: int | None = None) -> Seq[tuple[T, int]]

Return the n most common elements and their counts.

If n is None, then all elements are returned.

Parameters:

Name	Type	Description	Default
`n`	`int \| None`	Number of most common elements to return. Defaults to None (all elements).	`None`

Example:

>>> import pyochain as pc
>>> pc.Iter.from_([1, 1, 2, 3, 3, 3]).most_common(2).unwrap()
[(3, 3), (1, 2)]

Source code in src/pyochain/_iter/_eager.py

def most_common(self, n: int | None = None) -> Seq[tuple[T, int]]:
    """
    Return the n most common elements and their counts.

    If n is None, then all elements are returned.

    Args:
        n: Number of most common elements to return. Defaults to None (all elements).
    Example:
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([1, 1, 2, 3, 3, 3]).most_common(2).unwrap()
    [(3, 3), (1, 2)]

    ```
    """
    from collections import Counter

    from ._main import Seq

    def _most_common(data: Iterable[T]) -> list[tuple[T, int]]:
        return Counter(data).most_common(n)

    return Seq(self.into(_most_common))

pipe ¶

pipe(func: Callable[Concatenate[Self, P], R], *args: P.args, **kwargs: P.kwargs) -> R

Pipe the instance in the function and return the result.

Source code in src/pyochain/_core/_main.py

def pipe[**P, R](
    self,
    func: Callable[Concatenate[Self, P], R],
    *args: P.args,
    **kwargs: P.kwargs,
) -> R:
    """Pipe the instance in the function and return the result."""
    return func(self, *args, **kwargs)

println ¶

println(pretty: bool = True) -> Self

Print the underlying data and return self for chaining.

Useful for debugging, simply insert .println() in the chain, and then removing it will not affect the rest of the chain.

Source code in src/pyochain/_core/_main.py

def println(self, pretty: bool = True) -> Self:
    """
    Print the underlying data and return self for chaining.

    Useful for debugging, simply insert `.println()` in the chain,
    and then removing it will not affect the rest of the chain.
    """
    from pprint import pprint

    if pretty:
        pprint(self.unwrap(), sort_dicts=False)
    else:
        print(self.unwrap())
    return self

reduce ¶

reduce(func: Callable[[T, T], T]) -> T

Apply a function of two arguments cumulatively to the items of an iterable, from left to right.

Parameters:

Name	Type	Description	Default
`func`	`Callable[[T, T], T]`	Function to apply cumulatively to the items of the iterable.	required

This effectively reduces the iterable to a single value.

If initial is present, it is placed before the items of the iterable in the calculation.

It then serves as a default when the iterable is empty.

>>> import pyochain as pc
>>> pc.Iter.from_([1, 2, 3]).reduce(lambda a, b: a + b)
6

Source code in src/pyochain/_iter/_aggregations.py

def reduce(self, func: Callable[[T, T], T]) -> T:
    """
    Apply a function of two arguments cumulatively to the items of an iterable, from left to right.

    Args:
        func: Function to apply cumulatively to the items of the iterable.

    This effectively reduces the iterable to a single value.

    If initial is present, it is placed before the items of the iterable in the calculation.

    It then serves as a default when the iterable is empty.
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([1, 2, 3]).reduce(lambda a, b: a + b)
    6

    ```
    """
    return self.into(functools.partial(functools.reduce, func))

second ¶

second() -> T

Return the second element.

>>> import pyochain as pc
>>> pc.Iter.from_([9, 8]).second()
8

Source code in src/pyochain/_iter/_aggregations.py

def second(self) -> T:
    """
    Return the second element.
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([9, 8]).second()
    8

    ```
    """
    return self.into(cz.itertoolz.second)

sort ¶

sort(reverse: bool = False, key: Callable[[U], Any] | None = None) -> Seq[U]

Sort the elements of the sequence.

Note

This method must consume the entire iterable to perform the sort. The result is a new iterable over the sorted sequence.

Parameters:

Name	Type	Description	Default
`reverse`	`bool`	Whether to sort in descending order. Defaults to False.	`False`
`key`	`Callable[[U], Any] \| None`	Function to extract a comparison key from each element. Defaults to None.	`None`

Example:

>>> import pyochain as pc
>>> pc.Iter.from_([3, 1, 2]).sort().into(list)
[1, 2, 3]

Source code in src/pyochain/_iter/_eager.py

def sort[U: SupportsRichComparison[Any]](
    self: BaseEager[U], reverse: bool = False, key: Callable[[U], Any] | None = None
) -> Seq[U]:
    """
    Sort the elements of the sequence.

    Note:
        This method must consume the entire iterable to perform the sort.
        The result is a new iterable over the sorted sequence.

    Args:
        reverse: Whether to sort in descending order. Defaults to False.
        key: Function to extract a comparison key from each element. Defaults to None.
    Example:
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([3, 1, 2]).sort().into(list)
    [1, 2, 3]

    ```
    """

    def _sort(data: Iterable[U]) -> list[U]:
        return sorted(data, reverse=reverse, key=key)

    return self.collect(_sort)

stdev ¶

stdev() -> float

Return the standard deviation of the sequence.

>>> import pyochain as pc
>>> pc.Iter.from_([1, 2, 3]).stdev()
1.0

Source code in src/pyochain/_iter/_aggregations.py

def stdev[U: int | float](
    self: IterWrapper[U],
) -> float:
    """
    Return the standard deviation of the sequence.
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([1, 2, 3]).stdev()
    1.0

    ```
    """
    return self.into(statistics.stdev)

sum ¶

sum() -> U | Literal[0]

Return the sum of the sequence.

>>> import pyochain as pc
>>> pc.Iter.from_([1, 2, 3]).sum()
6

Source code in src/pyochain/_iter/_aggregations.py

def sum[U: int | float](self: IterWrapper[U]) -> U | Literal[0]:
    """
    Return the sum of the sequence.
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([1, 2, 3]).sum()
    6

    ```
    """
    return self.into(sum)

tail ¶

tail(n: int) -> Seq[T]

Return a tuple of the last n elements.

Parameters:

Name	Type	Description	Default
`n`	`int`	Number of elements to return.	required

Example:

>>> import pyochain as pc
>>> pc.Iter.from_([1, 2, 3]).tail(2).unwrap()
(2, 3)

Source code in src/pyochain/_iter/_eager.py

def tail(self, n: int) -> Seq[T]:
    """
    Return a tuple of the last n elements.

    Args:
        n: Number of elements to return.
    Example:
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([1, 2, 3]).tail(2).unwrap()
    (2, 3)

    ```
    """
    return self.collect(partial(cz.itertoolz.tail, n))

top_n ¶

top_n(n: int, key: Callable[[T], Any] | None = None) -> Seq[T]

Return a tuple of the top-n items according to key.

Parameters:

Name	Type	Description	Default
`n`	`int`	Number of top elements to return.	required
`key`	`Callable[[T], Any] \| None`	Function to extract a comparison key from each element. Defaults to None.	`None`

Example:

>>> import pyochain as pc
>>> pc.Iter.from_([1, 3, 2]).top_n(2).unwrap()
(3, 2)

Source code in src/pyochain/_iter/_eager.py

def top_n(self, n: int, key: Callable[[T], Any] | None = None) -> Seq[T]:
    """
    Return a tuple of the top-n items according to key.

    Args:
        n: Number of top elements to return.
        key: Function to extract a comparison key from each element. Defaults to None.
    Example:
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([1, 3, 2]).top_n(2).unwrap()
    (3, 2)

    ```
    """
    return self.collect(partial(cz.itertoolz.topk, n, key=key))

union ¶

union(*others: Iterable[T]) -> Seq[T]

Return the union of this iterable and 'others'.

Note

This method consumes inner data and removes duplicates.

Parameters:

Name	Type	Description	Default
`*others`	`Iterable[T]`	Other iterables to include in the union.	`()`

Example:

>>> import pyochain as pc
>>> pc.Iter.from_([1, 2, 2]).union([2, 3], [4]).iter().sort().unwrap()
[1, 2, 3, 4]

Source code in src/pyochain/_iter/_eager.py

def union(self, *others: Iterable[T]) -> Seq[T]:
    """
    Return the union of this iterable and 'others'.

    Note:
        This method consumes inner data and removes duplicates.

    Args:
        *others: Other iterables to include in the union.
    Example:
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([1, 2, 2]).union([2, 3], [4]).iter().sort().unwrap()
    [1, 2, 3, 4]

    ```
    """

    def _union(data: Iterable[T]) -> set[T]:
        return set(data).union(*others)

    return self.collect(_union)

unwrap ¶

unwrap() -> T

Return the underlying data.

This is a terminal operation.

Source code in src/pyochain/_core/_main.py

def unwrap(self) -> T:
    """
    Return the underlying data.

    This is a terminal operation.
    """
    return self._data

unzip ¶

unzip() -> Unzipped[U, V]

Converts an iterator of pairs into a pair of iterators.

Iter.unzip() consumes the iterator of pairs.

Returns an Unzipped NamedTuple, containing two iterators:

one from the left elements of the pairs

one from the right elements. This function is, in some sense, the opposite of zip.

>>> import pyochain as pc
>>> data = [(1, "a"), (2, "b"), (3, "c")]
>>> unzipped = pc.Iter.from_(data).unzip()
>>> unzipped.first.into(list)
[1, 2, 3]
>>> unzipped.second.into(list)
['a', 'b', 'c']

Source code in src/pyochain/_iter/_aggregations.py

def unzip[U, V](self: IterWrapper[tuple[U, V]]) -> Unzipped[U, V]:
    """
    Converts an iterator of pairs into a pair of iterators.

    `Iter.unzip()` consumes the iterator of pairs.

    Returns an Unzipped NamedTuple, containing two iterators:

    - one from the left elements of the pairs
    - one from the right elements.
    This function is, in some sense, the opposite of zip.
    ```python
    >>> import pyochain as pc
    >>> data = [(1, "a"), (2, "b"), (3, "c")]
    >>> unzipped = pc.Iter.from_(data).unzip()
    >>> unzipped.first.into(list)
    [1, 2, 3]
    >>> unzipped.second.into(list)
    ['a', 'b', 'c']

    ```
    """
    from ._main import Iter

    def _unzip(data: Iterable[tuple[U, V]]) -> Unzipped[U, V]:
        d: list[tuple[U, V]] = list(data)
        return Unzipped(Iter(x[0] for x in d), Iter(x[1] for x in d))

    return self.into(_unzip)

variance ¶

variance() -> float

Return the variance of the sequence.

>>> import pyochain as pc
>>> pc.Iter.from_([1, 2, 3, 7, 8]).variance()
9.7

Source code in src/pyochain/_iter/_aggregations.py

def variance[U: int | float](
    self: IterWrapper[U],
) -> float:
    """
    Return the variance of the sequence.
    ```python
    >>> import pyochain as pc
    >>> pc.Iter.from_([1, 2, 3, 7, 8]).variance()
    9.7

    ```
    """
    return self.into(statistics.variance)

Seq¶

__slots__ class-attribute instance-attribute ¶

_data instance-attribute ¶

__init__ ¶

apply ¶

argmax ¶

argmin ¶

collect ¶

combination_index ¶

count ¶

diff_symmetric ¶

diff_unique ¶

first ¶

intersection ¶

into ¶

item ¶

iter ¶

last ¶

max ¶

mean ¶

median ¶

min ¶

mode ¶

most_common ¶

pipe ¶

println ¶

reduce ¶

second ¶

sort ¶

stdev ¶

sum ¶

tail ¶

top_n ¶

union ¶

unwrap ¶

unzip ¶

variance ¶

slots `class-attribute` `instance-attribute` ¶

_data `instance-attribute` ¶

init ¶