listless ⚓︎

Listless = generators, iterators, and async iterators, Oh My!

Functions:

aiterable –

Convert any-iterable to an async iterator
chunk –

Yield chunks of size n from an iterable/sequence/collection
chunks –

Yield chunks of something slice-able with length <= chunk_size
chunkseq –

Yield chunks of size n from a Sequence
chunkstr –

Yield chunks of size n from a string
enumerate_async –

Enumerate (async) over any iterable
exhaust –

Exhaust an iterable; useful for evaluating a map object.
filter_is_none –

Filter values that is None; checkout filter_none for false-y filtering
filter_none –

Filter None values from an iterable
flatten –

Flatten possibly nested iterables of sequences to a flat list
flatten_seq –

Flatten possibly nested iterables of sequences to a flat list
flatten_strings –

Flatten possibly nested iterables of sequences to a list of strings
is_sequence –

Check if an object is a sequence
it_product –

Product of all the elements in an iterable of numbers
itlen –

Return the length/num-items in an iterable
list_async –

Consume any iterable (async/sync) and return as a list
next_async –

Return the next item of any iterator/iterable (sync or async
nyield –

Yield the first n items of an iterable
pairs –

Yield pairs of adjacent elements
partition –

Partition an iterable into chunks of size n
set_async –

Consume any iterable (async/sync) and return as a list
spliterable –

1 generator + True/False-function => 2 generators (True-gen, False-gen)
unique –

Alias for unique_gen
unique_gen –

Yield unique values (ordered) from an iterable
xmap –

Apply a function to each element of an iterable immediately
zip_async –

Async version of builtin zip function

aiterable ⚓︎

aiterable(
    it: Iterable[_T] | AsyncIterable[_T],
) -> AsyncIterator[_T]

Convert any-iterable to an async iterator

Examples:

>>> from asyncio import run
>>> plain_jane_list = list(range(10))
>>> async def consume_aiterable(it):
...     stuff = []
...     async for el in aiterable(it):
...         stuff.append(el)
...     return stuff
>>> run(consume_aiterable(plain_jane_list))
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
>>> async def async_gen():
...     for b in range(10):
...        yield b
>>> run(consume_aiterable(async_gen()))
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
>>> class AsyncIterable:
...     def __aiter__(self):
...         return async_gen()
>>> run(consume_aiterable(AsyncIterable()))
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

chunk ⚓︎

chunk(
    it: Sequence[_T" backlink-type="used-by" backlink-anchor="listless.chunk" optional hover>_T] | Collection[_T" backlink-type="used-by" backlink-anchor="listless.chunk" optional hover>_T], n: int
) -> Iterable[Sequence[_T" backlink-type="returned-by" backlink-anchor="listless.chunk" optional hover>_T] | Collection[_T] | str]

Yield chunks of size n from an iterable/sequence/collection

Examples:

>>> list(chunk([1, 2, 3, 4, 5, 6], 3))
[[1, 2, 3], [4, 5, 6]]

chunks ⚓︎

chunks(it: str, chunk_size: int) -> Iterable[str]

chunks(it: list[_T], chunk_size: int) -> Iterable[list[_T]]

chunks(
    it: Sequence[_T], chunk_size: int
) -> Iterable[Sequence[_T]]

chunks(
    it: Collection[_T], chunk_size: int
) -> Iterable[Collection[_T]]

chunks(
    it: Sequence[_T] | Collection[_T], chunk_size: int
) -> Iterable[Sequence[_T] | Collection[_T] | str]

Yield chunks of something slice-able with length <= chunk_size

Parameters:

it ⚓︎
(Iterable[Any]) –

Iterable to chunk
chunk_size ⚓︎
(int) –

Size of the chunks

Returns:

Iterable[Sequence[_T] | Collection[_T] | str] –

Iterable of the chunks

Examples:

>>> list(chunks([1, 2, 3, 4, 5, 6], 3))
[[1, 2, 3], [4, 5, 6]]
>>> list(chunks([1, 2, 3, 4, 5, 6], 2))
[[1, 2], [3, 4], [5, 6]]
>>> list(chunks('abcdefghijklmnopqrstuvwxyz', 13))
['abcdefghijklm', 'nopqrstuvwxyz']

Can chunk where it length is not divisible by chunk_size

>>> list(chunks('abcdefghijklmnopqrstuvwxyz', 4))
['abcd', 'efgh', 'ijkl', 'mnop', 'qrst', 'uvwx', 'yz']

>>> list(chunks((el for el in range(10)), 4))
[(0, 1, 2, 3), (4, 5, 6, 7), (8, 9)]

chunkseq ⚓︎

chunkseq(
    it: Sequence[_T], n: int
) -> Iterable[Sequence[_T]]

Yield chunks of size n from a Sequence

chunkstr ⚓︎

chunkstr(string: str, n: int) -> Iterable[str]

Yield chunks of size n from a string

enumerate_async `async` ⚓︎

enumerate_async(
    it: AnyIterable[_T], start: int = 0
) -> AsyncIterator[tuple[int, _T]]

Enumerate (async) over any iterable

Examples:

>>> async def t():
...     return [item async for item in enumerate_async('abcde')]
>>> from asyncio import run as aiorun
>>> aiorun(t())
[(0, 'a'), (1, 'b'), (2, 'c'), (3, 'd'), (4, 'e')]

exhaust ⚓︎

exhaust(
    it: Iterable[_T], *, maxlen: int | None = 0
) -> deque[_T]

Exhaust an iterable; useful for evaluating a map object.

Parameters:

it ⚓︎
(Iterable[_T]) –

Iterable to exhaust
maxlen ⚓︎
(int | None, default: 0 ) –

Maximum length of the deque; if 0, deque is unbounded

Examples:

>>> a = [1, 2, 3, 4, 5, 6]
>>> a_map = map(lambda x: x*2, a)
>>> a_exhausted = exhaust(a_map)
>>> a = [1, 2, 3, 4, 5, 6]
>>> b = []
>>> def square_and_append_to_b(n):
...     b.append(n**2)
>>> a_map = map(square_and_append_to_b, a)
>>> a_exhausted = exhaust(a_map)
>>> a_exhausted
deque([], maxlen=0)
>>> b
[1, 4, 9, 16, 25, 36]
>>> another_map = map(lambda x: x*2, a)
>>> another_exhausted = exhaust(another_map, maxlen=2)
>>> another_exhausted
deque([10, 12], maxlen=2)

filter_is_none ⚓︎

filter_is_none(it: Iterable[_T | None]) -> Iterable[_T]

Filter values that is None; checkout filter_none for false-y filtering

Parameters:

it ⚓︎
(Iterable[_T | None]) –

Iterable possibly containing None/False-y values

Returns:

Iterable[_T] –

filter object with None values excluded

Examples:

>>> list(filter_is_none([1, 2, None, 3, 4, None, 5, "a_string???"]))
[1, 2, 3, 4, 5, 'a_string???']
>>> list(filter_is_none([-1, 0, 1, '', 's', None, [], ['s'], {}]))
[-1, 0, 1, '', 's', [], ['s'], {}]

filter_none ⚓︎

filter_none(it: Iterable[_T | None]) -> Iterable[_T]

Filter None values from an iterable

Parameters:

it ⚓︎
(Iterable[_T | None]) –

Iterable possibly containing None values

Returns:

Iterable[_T] –

filter object with None values excluded

Examples:

>>> list(filter_none([1, 2, None, 3, 4, None, 5, "a_string???"]))
[1, 2, 3, 4, 5, 'a_string???']
>>> list(filter_none([1, 2, '', 3, 4, None, 5, "a_string???", []]))
[1, 2, 3, 4, 5, 'a_string???']
>>> list(filter_none([-1, 0, 1, '', 's', None, [], ['s'], {}]))
[-1, 1, 's', ['s']]

This function is p simple and importing it and calling it might actually be more characters to type than just using filter(None, ya_iterable) but it is a fire thing to know and you can totally show off with this, also outperforms the list/gen comprehension equivalent, by quite a bit::

import random
res = [random.randrange(1, 300, 1) for i in range(1000)]

lists = [
    res
]
for i in range(40):
    random.shuffle(res)
    lists.append(res)

def filter_one(it):
    return (i for i in it if i is not None)

def filter_two(it):
    return filter(None, it)

Timing the first function (generator comprehension)::

%%timeit
for i in range(100):
    for l in lists:
        a = list(filter_one(l))

180 ms +/- 184 μs/loop (mean +/- std. dev. of 7 runs, 10 loops each)

Timing the second function (filter)::

%%timeit
for i in range(100):
    for l in lists:
        a = list(filter_two(l))

42.5 ms +/- 112 μs/loop (mean +/- std. dev. of 7 runs, 10 loops each)

flatten ⚓︎

flatten(*args: _T | Iterable[_T]) -> Iterable[_T]

Flatten possibly nested iterables of sequences to a flat list

Examples:

>>> list(flatten("cmd", ["uno", "dos", "tres"]))
['cmd', 'uno', 'dos', 'tres']
>>> list(flatten("cmd", ["uno", "dos", "tres", ["4444", "five"]]))
['cmd', 'uno', 'dos', 'tres', '4444', 'five']
>>> list(flatten("cmd", ["uno", "dos", "tres", ["4444", "five", 123]]))
['cmd', 'uno', 'dos', 'tres', '4444', 'five', 123]

flatten_seq ⚓︎

flatten_seq(
    *args: _T | Sequence[_T], anystr: bool = False
) -> Iterable[_T]

Flatten possibly nested iterables of sequences to a flat list

Examples:

>>> list(flatten_seq("cmd", ["uno", "dos", "tres"]))
['cmd', 'uno', 'dos', 'tres']
>>> list(flatten_seq("cmd", ["uno", "dos", "tres", ["4444", "five"]]))
['cmd', 'uno', 'dos', 'tres', '4444', 'five']

flatten_strings ⚓︎

flatten_strings(
    *args: Any | Iterable[Any],
) -> Iterable[str]

Flatten possibly nested iterables of sequences to a list of strings

Examples:

>>> from listless import flatten_strings
>>> list(flatten_strings("cmd", ["uno", "dos", "tres"]))
['cmd', 'uno', 'dos', 'tres']
>>> list(flatten_strings("cmd", ["uno", "dos", "tres", ["4444", "five", 123]]))
['cmd', 'uno', 'dos', 'tres', '4444', 'five', '123']

is_sequence ⚓︎

is_sequence(seq: Any) -> bool

Check if an object is a sequence

Examples:

>>> is_sequence([1, 2, 3])
True
>>> is_sequence('abc')
False
>>> is_sequence(1)
False
>>> is_sequence(None)
False
>>> is_sequence(True)
False
>>> is_sequence((1, 2, 3))
True

is_subscriptable ⚓︎

is_subscriptable(obj: Any) -> bool

Check if an object is subscriptable

Examples:

>>> is_subscriptable([1, 2, 3])
True
>>> is_subscriptable('abc')
True
>>> is_subscriptable(1)
False
>>> is_subscriptable(None)
False
>>> is_subscriptable(True)
False
>>> is_subscriptable((1, 2, 3))
True

it_product ⚓︎

it_product(it: Iterable[int | float]) -> int | float

Product of all the elements in an iterable of numbers

Parameters:

it ⚓︎
(Iterable[int | float]) –

Iterable of numbers

Returns:

int | float –

The product of all the numbers in the iterable

Examples:

>>> it_product([1, 2, 3, 4])
24
>>> it_product(tuple([1, 2, 3, 4]))
24
>>> it_product([-1, -2, -3, 4])
-24
>>> it_product([-1, -2, 3, 4, 0.5])
12.0

itlen ⚓︎

itlen(
    iterable: Iterable[Any], *, unique: bool = False
) -> int

Return the length/num-items in an iterable

This consumes the iterable.

Parameters:

iterable ⚓︎
(Iterable[Any]) –

Iterable
unique ⚓︎
(bool, default: False ) –

Count unique values

Returns:

int –

Length of an iterable

Examples:

>>> itlen(range(10))
10
>>> itlen(x for x in range(1000000) if x % 3 == 0)
333334
>>> l = [x for x in range(1000000) if x % 3 == 0]
>>> itlen(l + l, unique=True)
333334

list_async `async` ⚓︎

list_async(itr: AnyIterable[_T]) -> list[_T]

Consume any iterable (async/sync) and return as a list

Examples:

>>> async def t():
...     return await list_async(range(5))
>>> from asyncio import run as aiorun
>>> aiorun(t())
[0, 1, 2, 3, 4]

next_async `async` ⚓︎

next_async(it: AnyIterator[_T]) -> _T

Return the next item of any iterator/iterable (sync or async

Examples:

>>> from asyncio import run as aiorun
>>> async def async_gen():
...     for b in range(10):
...        yield b
>>> gen = async_gen()
>>> async def fn(gen):
...     first = await next_async(gen)
...     second = await next_async(gen)
...     return first, second
>>> aiorun(fn(gen))
(0, 1)
>>> aiorun(fn(iter(range(2))))
(0, 1)

nyield ⚓︎

nyield(it: Sequence[_T], n: int) -> Iterable[_T]

Yield the first n items of an iterable

Examples:

>>> list(nyield([1, 2, 3, 4, 5, 6], 3))
[1, 2, 3]

pairs ⚓︎

pairs(it: Iterable[_T]) -> Iterable[tuple[_T, _T]]

Yield pairs of adjacent elements

Examples:

>>> list(pairs([1, 2, 3, 4]))
[(1, 2), (2, 3), (3, 4)]
>>> list(pairs(['a', 'b', 'c']))
[('a', 'b'), ('b', 'c')]
>>> list(pairs('abc'))
[('a', 'b'), ('b', 'c')]

partition ⚓︎

partition(
    it: Sequence[it" backlink-type="used-by" backlink-anchor="listless.partition" optional hover>it],
    n: int,
    *,
    pad: bool = False,
    padval: Any = None,
) -> Iterable[Sequence[_T]]

Partition an iterable into chunks of size n

Parameters:

it ⚓︎
(Sequence[_T]) –

Iterable to partition
n ⚓︎
(int) –

Size of the partition chunks
pad ⚓︎
(bool, default: False ) –

Pad parts with padval if True, else do not pad
padval ⚓︎
(Any, default: None ) –

Value to pad with

Returns:

Iterable[Sequence[_T]] –

Iterable of the partitioned chunks

Examples:

>>> list(partition([1, 2, 3, 4, 5, 6], 3))
[(1, 2, 3), (4, 5, 6)]
>>> list(partition([1, 2, 3, 4, 5, 6], 2))
[(1, 2), (3, 4), (5, 6)]
>>> for part in partition('abcdefghijklmnopqrstuvwxyz', 13):
...    print(part)
('a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm')
('n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z')
>>> for part in partition('abcdefghijklmnopqrstuvwxyz', 4):
...    print(part)
('a', 'b', 'c', 'd')
('e', 'f', 'g', 'h')
('i', 'j', 'k', 'l')
('m', 'n', 'o', 'p')
('q', 'r', 's', 't')
('u', 'v', 'w', 'x')
>>> for part in partition('abcdefghijklmnopqrstuvwxyz', 4, pad=True):
...    print(part)
('a', 'b', 'c', 'd')
('e', 'f', 'g', 'h')
('i', 'j', 'k', 'l')
('m', 'n', 'o', 'p')
('q', 'r', 's', 't')
('u', 'v', 'w', 'x')
('y', 'z', None, None)
>>> for part in partition('abcdefghijklmnopqrstuvwxyz', 4, pad=True, padval=...):
...   print(part)
('a', 'b', 'c', 'd')
('e', 'f', 'g', 'h')
('i', 'j', 'k', 'l')
('m', 'n', 'o', 'p')
('q', 'r', 's', 't')
('u', 'v', 'w', 'x')
('y', 'z', Ellipsis, Ellipsis)

Raises:

TypeError –

If n is not and int
ValueError –

if n is less than 1

set_async `async` ⚓︎

set_async(itr: AnyIterable[_T]) -> set[_T]

Consume any iterable (async/sync) and return as a list

Examples:

>>> async def t():
...     return await set_async(range(5))
>>> from asyncio import run as aiorun
>>> aiorun(t())
{0, 1, 2, 3, 4}

spliterable ⚓︎

spliterable(
    it: Iterable[_T], fn: Callable[[_T], bool]
) -> tuple[Iterable[_T], Iterable[_T]]

1 generator + True/False-function => 2 generators (True-gen, False-gen)

Parameters:

it ⚓︎
(Iterable[_T]) –

iterable to split
fn ⚓︎
(Callable[[_T], bool]) –

Function to evaluate iterable elements and returns True or False

Returns:

tuple[Iterable[_T], Iterable[_T]] –

tuple of generators: (gen_predicate_true, gen_predicate_false)

Examples:

>>> is_even = lambda n: n % 2 == 0
>>> a, b = spliterable(range(10), is_even)
>>> list(a)
[0, 2, 4, 6, 8]
>>> list(b)
[1, 3, 5, 7, 9]

unique ⚓︎

unique(
    it: Iterable[_T], key: Callable[[_T], _K] | None = None
) -> Iterable[_T]

Alias for unique_gen

Examples:

>>> l = [*range(10), *range(10)]
>>> list(unique(l))
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
>>> tuple(unique(['cat', 'mouse', 'dog', 'hen'], key=len))
('cat', 'mouse')

unique_gen ⚓︎

unique_gen(
    it: Iterable[_T], key: Callable[[_T], _K] | None = None
) -> Iterable[_T]

Yield unique values (ordered) from an iterable

Parameters:

it ⚓︎
(Iterable[_T]) –

Iterable
key ⚓︎
(Callable[[_T], _K] | None, default: None ) –

Optional callable to use to get the key to use for uniqueness

Returns:

Iterable[_T] –

Generator that yields unique values as they appear

Examples:

>>> l = [*range(10), *range(10)]
>>> list(unique_gen(l))
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
>>> tuple(unique_gen(['cat', 'mouse', 'dog', 'hen'], key=len))
('cat', 'mouse')

xmap ⚓︎

xmap(
    func: Callable[[_T], _R],
    it: Iterable[_T],
    *,
    maxlen: int | None = 0,
) -> deque[_R]

Apply a function to each element of an iterable immediately

Parameters:

func ⚓︎
(Callable[[_T], _R]) –

Function to apply to each element
it ⚓︎
(Iterable[_T]) –

iterable to apply func to
maxlen ⚓︎
(int | None, default: 0 ) –

Maximum length of the deque; if 0, deque is unbounded

Returns:

deque[_R] –

Deque of the possible results if maxlen is greater than 0

Examples:

>>> xmap(lambda x: x*2, list(range(1, 7)))
deque([], maxlen=0)
>>> xmap(lambda x: x*2, list(range(1, 7)), maxlen=2)
deque([10, 12], maxlen=2)
>>> xmap(lambda x: x*2, list(range(1, 7)), maxlen=None)
deque([2, 4, 6, 8, 10, 12])

zip_async `async` ⚓︎

zip_async(
    *iterables: AnyIterable[Any],
) -> AsyncIterator[tuple[Any, ...]]

Async version of builtin zip function

Examples:

>>> from asyncio import run as aiorun
>>> from listless import zip_async
>>> from listless import list_async, iter_async  # for fake async iters
>>> a, b, c = iter_async(range(4)), iter_async(range(6)), iter_async(range(5))
>>> aiorun(list_async(zip_async(a, b, c)))
[(0, 0, 0), (1, 1, 1), (2, 2, 2), (3, 3, 3)]

listless ⚓︎

aiterable ⚓︎

chunk ⚓︎

chunks ⚓︎

it ⚓︎

chunk_size ⚓︎

chunkseq ⚓︎

chunkstr ⚓︎

enumerate_async async ⚓︎

exhaust ⚓︎

it ⚓︎

maxlen ⚓︎

filter_is_none ⚓︎

it ⚓︎

filter_none ⚓︎

it ⚓︎

flatten ⚓︎

flatten_seq ⚓︎

flatten_strings ⚓︎

is_sequence ⚓︎

is_subscriptable ⚓︎

it_product ⚓︎

it ⚓︎

itlen ⚓︎

iterable ⚓︎

unique ⚓︎

list_async async ⚓︎

next_async async ⚓︎

nyield ⚓︎

pairs ⚓︎

partition ⚓︎

it ⚓︎

n ⚓︎

pad ⚓︎

padval ⚓︎

set_async async ⚓︎

spliterable ⚓︎

it ⚓︎

fn ⚓︎

unique ⚓︎

unique_gen ⚓︎

it ⚓︎

key ⚓︎

xmap ⚓︎

func ⚓︎

it ⚓︎

maxlen ⚓︎

zip_async async ⚓︎

`it` ⚓︎

`chunk_size` ⚓︎

enumerate_async `async` ⚓︎

`it` ⚓︎

`maxlen` ⚓︎

`it` ⚓︎

`it` ⚓︎

`it` ⚓︎

`iterable` ⚓︎

`unique` ⚓︎

list_async `async` ⚓︎

next_async `async` ⚓︎

`it` ⚓︎

`n` ⚓︎

`pad` ⚓︎

`padval` ⚓︎

set_async `async` ⚓︎

`it` ⚓︎

`fn` ⚓︎

`it` ⚓︎

`key` ⚓︎

`func` ⚓︎

`it` ⚓︎

`maxlen` ⚓︎

zip_async `async` ⚓︎