Skip to content

listless

listless ⚓︎

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

Functions⚓︎

listless.aiterable(it: Union[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]
Source code in libs/listless/listless/__init__.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def aiterable(it: Union[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]

    """
    if isinstance(it, AsyncIterator):
        return it

    if isinstance(it, AsyncIterable):
        return it.__aiter__()

    async def gen() -> AsyncIterator[_T]:
        for item in it:
            yield item

    return gen()

listless.chunk(it: Sequence[_T], n: int) -> Iterable[Sequence[_T]] ⚓︎

Yield chunks of size n from a Sequence

Examples:

>>> list(chunk([1, 2, 3, 4, 5, 6], 3))
[[1, 2, 3], [4, 5, 6]]
Source code in libs/listless/listless/__init__.py 
324
325
326
327
328
329
330
331
332
def chunk(it: Sequence[_T], n: int) -> Iterable[Sequence[_T]]:
    """Yield chunks of size n from a Sequence

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

    """
    return chunks(it, n)

listless.chunks(it: Sequence[_T], chunk_size: int) -> Iterable[Union[Sequence[_T], str]] ⚓︎

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

Parameters:

Name Type Description Default
it Iterable[Any]

Iterable to chunk

 required
chunk_size int

Size of the chunks

 required

Returns:

Type Description
Iterable[Union[Sequence[_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)]
Source code in libs/listless/listless/__init__.py 
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
def chunks(it: Sequence[_T], chunk_size: int) -> Iterable[Union[Sequence[_T], str]]:
    """Yield chunks of something slice-able with length <= chunk_size

    Args:
        it (Iterable[Any]): Iterable to chunk
        chunk_size (int): Size of the chunks

    Returns:
        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)]

    """
    if isinstance(it, str):
        yield from chunkstr(it, chunk_size)
    elif isinstance(it, (list, tuple)) or is_sequence(it):
        yield from chunkseq(it, chunk_size)
    else:
        while True:
            _chunk = tuple(islice(it, chunk_size))
            if not _chunk:
                break
            yield _chunk

listless.chunkseq(it: Sequence[_T], n: int) -> Iterable[Sequence[_T]] ⚓︎

Yield chunks of size n from a Sequence

Source code in libs/listless/listless/__init__.py 
263
264
265
def chunkseq(it: Sequence[_T], n: int) -> Iterable[Sequence[_T]]:
    """Yield chunks of size n from a Sequence"""
    return (it[i : i + n] for i in range(0, len(it), n))

listless.chunkstr(string: str, n: int) -> Iterable[str] ⚓︎

Yield chunks of size n from a string

Source code in libs/listless/listless/__init__.py 
268
269
270
def chunkstr(string: str, n: int) -> Iterable[str]:
    """Yield chunks of size n from a string"""
    return (string[i : i + n] for i in range(0, len(string), n))

listless.enumerate_async(it: AnyIterable[_T], start: int = 0) -> AsyncIterator[Tuple[int, _T]] async ⚓︎

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')]
Source code in libs/listless/listless/__init__.py 
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
async def 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')]

    """
    index = start
    async for item in aiterable(it):  # pragma: no cover
        yield index, item
        index += 1

listless.exhaust(it: Iterable[_T], *, maxlen: Optional[int] = 0) -> Deque[_T] ⚓︎

Exhaust an interable; useful for evaluating a map object.

Parameters:

Name Type Description Default
it Iterable[_T]

Iterable to exhaust

 required
maxlen Optional[int]

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

 0

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)
Source code in libs/listless/listless/__init__.py 
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
def exhaust(it: Iterable[_T], *, maxlen: Optional[int] = 0) -> Deque[_T]:
    """Exhaust an interable; useful for evaluating a map object.

    Args:
        it: Iterable to exhaust
        maxlen: 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)

    """
    return deque(it, maxlen=maxlen)

listless.filter_is_none(it: Iterable[Union[_T, None]]) -> Iterable[_T] ⚓︎

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

Parameters:

Name Type Description Default
it Iterable[Union[_T, None]]

Iterable possibly containing None/False-y values

 required

Returns:

Type Description
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'], {}]
Source code in libs/listless/listless/__init__.py 
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
def filter_is_none(it: Iterable[Union[_T, None]]) -> Iterable[_T]:
    """Filter values that `is None`; checkout filter_none for false-y filtering

    Args:
        it: Iterable possibly containing None/False-y values

    Returns:
        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'], {}]


    """
    return filter(lambda x: x is not None, it)  # type: ignore[arg-type]

listless.filter_none(it: Iterable[Union[_T, None]]) -> Iterable[_T] ⚓︎

Filter None values from an iterable

Parameters:

Name Type Description Default
it Iterable[Union[_T, None]]

Iterable possibly containing None values

 required

Returns:

Type Description
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)
Source code in libs/listless/listless/__init__.py 
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
def filter_none(it: Iterable[Union[_T, None]]) -> Iterable[_T]:
    """Filter `None` values from an iterable

    Args:
        it: Iterable possibly containing None values

    Returns:
        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)


    """
    return filter(None, it)

listless.flatten(*args: Union[_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]
Source code in libs/listless/listless/__init__.py 
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
def flatten(*args: Union[_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]

    """
    return cast(
        Iterable[_T],
        chain(
            *(
                flatten(*arg) if isinstance(arg, (list, tuple)) else (arg,)
                for arg in args
            )
        ),
    )

listless.flatten_seq(*args: Union[_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']
Source code in libs/listless/listless/__init__.py 
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
def flatten_seq(*args: Union[_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']

    """
    return list(
        reduce(
            iconcat,
            [
                flatten_seq(*arg) if isinstance(arg, (list, tuple)) else (arg,)
                for arg in args
            ],
            [],
        )
    )

listless.flatten_strings(*args: Union[_T, Iterable[_T]]) -> 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']
Source code in libs/listless/listless/__init__.py 
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
def flatten_strings(
    *args: Union[
        _T,
        Iterable[_T],
    ],
) -> 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']

    """
    return (str(arg) for arg in flatten(*args))

listless.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
Source code in libs/listless/listless/__init__.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
def 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

    """

    if isinstance(seq, str):
        return False
    try:
        len(seq)
    except TypeError:
        return False
    return is_subscriptable(seq)

listless.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
Source code in libs/listless/listless/__init__.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def 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

    """
    try:
        obj[0]
    except TypeError:
        return False
    return True

listless.it_product(it: Iterable[Union[int, float]]) -> Union[int, float] ⚓︎

Product of all the elements in an iterable of numbers

Parameters:

Name Type Description Default
it Iterable[Union[int, float]]

Iterable of numbers

 required

Returns:

Type Description
Union[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
Source code in libs/listless/listless/__init__.py 
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
def it_product(it: Iterable[Union[int, float]]) -> Union[int, float]:
    """Product of all the elements in an iterable of numbers

    Args:
        it: Iterable of numbers

    Returns:
        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

    """
    return reduce(mul, it)

listless.itlen(iterable: Iterable[Any], unique: bool = False) -> int ⚓︎

Return the length/num-items in an iterable

This consumes the iterable.

Parameters:

Name Type Description Default
iterable Iterable[Any]

Iterable

 required
unique bool

Count unique values

 False

Returns:

Type Description
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
Source code in libs/listless/listless/__init__.py 
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
def itlen(iterable: Iterable[Any], unique: bool = False) -> int:
    """Return the length/num-items in an iterable

    This consumes the iterable.

    Args:
        iterable: Iterable
        unique: Count unique values

    Returns:
        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

    """
    if unique:
        return len(set(iterable))
    counter = count()
    deque(zip(iterable, counter), maxlen=0)
    return next(counter)

listless.list_async(itr: AnyIterable[_T]) -> List[_T] async ⚓︎

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]
Source code in libs/listless/listless/__init__.py 
695
696
697
698
699
700
701
702
703
704
705
706
async def 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]

    """
    return [item async for item in aiterable(itr)]

listless.next_async(it: AnyIterator[_T]) -> _T async ⚓︎

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)
Source code in libs/listless/listless/__init__.py 
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
async def 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)

    """
    if isinstance(it, AsyncIterator):
        return await it.__anext__()

    try:
        return next(it)
    except StopIteration:  # pragma: no cover
        raise StopAsyncIteration  # noqa: B904

listless.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]
Source code in libs/listless/listless/__init__.py 
199
200
201
202
203
204
205
206
207
def 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]

    """
    return islice(it, n)

listless.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')]
Source code in libs/listless/listless/__init__.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def 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')]

    """
    a, b = tee(it)
    next(b, None)
    return zip(a, b)

listless.partition(it: Sequence[_T], n: int, *, pad: bool = False, padval: Any = None) -> Iterable[Sequence[_T]] ⚓︎

Partition an iterable into chunks of size n

Parameters:

Name Type Description Default
it Sequence[_T]

Iterable to partition

 required
n int

Size of the partition chunks

 required
pad bool

Pad parts with padval if True, else do not pad

 False
padval Any

Value to pad with

 None

Returns:

Type Description
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:

Type Description
TypeError

If n is not and int
ValueError

if n is less than 1
Source code in libs/listless/listless/__init__.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def partition(
    it: Sequence[_T],
    n: int,
    *,
    pad: bool = False,
    padval: Any = None,
) -> Iterable[Sequence[_T]]:
    """Partition an iterable into chunks of size n

    Args:
        it: Iterable to partition
        n (int): Size of the partition chunks
        pad (bool): Pad parts with padval if True, else do not pad
        padval (Any): Value to pad with

    Returns:
        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

    """
    if not isinstance(n, int):
        raise TypeError("n must be an integer")  # pragma: no cover
    if n < 1:
        raise ValueError("n must be >= 1")  # pragma: no cover
    args = [iter(it)] * n
    if pad:
        return zip_longest(*args, fillvalue=padval)
    else:
        return zip(*args)

listless.set_async(itr: AnyIterable[_T]) -> Set[_T] async ⚓︎

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}
Source code in libs/listless/listless/__init__.py 
709
710
711
712
713
714
715
716
717
718
719
720
async def 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}

    """
    return {item async for item in aiterable(itr)}

listless.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:

Name Type Description Default
it Iterable[_T]

iterable to split

 required
fn Callable[[_T], bool]

Function to evaluate iterable elements and returns True or False

 required

Returns:

Type Description
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]
Source code in libs/listless/listless/__init__.py 
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
def spliterable(
    it: Iterable[_T], fn: Callable[[_T], bool]
) -> Tuple[Iterable[_T], Iterable[_T]]:
    """1 generator + True/False-function => 2 generators (True-gen, False-gen)

    Args:
        it: iterable to split
        fn: Function to evaluate iterable elements and returns True or False

    Returns:
        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]

    """
    _true_gen, _false_gen = tee((fn(item), item) for item in it)
    return (i for p, i in _true_gen if p), (i for p, i in _false_gen if not p)

listless.unique(it: Iterable[_T], key: Optional[Callable[[_T], _K]] = 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')
Source code in libs/listless/listless/__init__.py 
648
649
650
651
652
653
654
655
656
657
658
659
def unique(it: Iterable[_T], key: Optional[Callable[[_T], _K]] = 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')

    """
    return unique_gen(it=it, key=key)

listless.unique_gen(it: Iterable[_T], key: Optional[Callable[[_T], _K]] = None) -> Iterable[_T] ⚓︎

Yield unique values (ordered) from an iterable

Parameters:

Name Type Description Default
it Iterable[_T]

Iterable

 required
key Optional[Callable[[_T], _K]]

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

 None

Returns:

Type Description
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')
Source code in libs/listless/listless/__init__.py 
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
def unique_gen(
    it: Iterable[_T], key: Optional[Callable[[_T], _K]] = None
) -> Iterable[_T]:
    """Yield unique values (ordered) from an iterable

    Args:
        it: Iterable
        key: Optional callable to use to get the key to use for uniqueness

    Returns:
        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')

    """
    if key is None:
        have: Set[_T] = set()
        have_add = have.add
        return (x for x in it if not (x in have or have_add(x)))
    else:
        havek: Set[_K] = set()
        havek_add = havek.add

        return (
            el
            for el, k_el in ((_el, key(_el)) for _el in it)
            if not (k_el in havek or havek_add(k_el))
        )

listless.xmap(func: Callable[[_T], _R], it: Iterable[_T], *, maxlen: Optional[int] = 0) -> Deque[_R] ⚓︎

Apply a function to each element of an iterable immediately

Parameters:

Name Type Description Default
func Callable[[_T], _R]

Function to apply to each element

 required
it Iterable[_T]

iterable to apply func to

 required
maxlen Optional[int]

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

 0

Returns:

Type Description
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])
Source code in libs/listless/listless/__init__.py 
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
def xmap(
    func: Callable[[_T], _R], it: Iterable[_T], *, maxlen: Optional[int] = 0
) -> Deque[_R]:
    """Apply a function to each element of an iterable immediately

    Args:
        func: Function to apply to each element
        it: iterable to apply func to
        maxlen: Maximum length of the deque; if 0, deque is unbounded

    Returns:
        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])

    """
    return exhaust(map(func, it), maxlen=maxlen)

listless.zip_async(*iterables: AnyIterable[Any]) -> AsyncIterator[Tuple[Any, ...]] async ⚓︎

Async verstion of builtin zip function

Example

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)]

Source code in libs/listless/listless/__init__.py 
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
async def zip_async(*iterables: AnyIterable[Any]) -> AsyncIterator[Tuple[Any, ...]]:
    """Async verstion of builtin zip function

    Example:
        >>> 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)]

    """
    its: List[AsyncIterator[Any]] = [aiterable(it) for it in iterables]

    while True:
        try:
            values = await asyncio.gather(*[it.__anext__() for it in its])
            yield tuple(values)
        except (StopIteration, StopAsyncIteration):
            break

Modules⚓︎

listless.__about__ ⚓︎

Package metadata/info

listless.__main__ ⚓︎

pkg entry ~ python -m listless