Source code for stdlibx.itertools._iterable

 1from __future__ import annotations
 2
 3import functools
 4from typing import TYPE_CHECKING, Callable, Iterable, TypeVar, Union
 5
 6if TYPE_CHECKING:
 7    from stdlibx.itertools._types import Operation
 8
 9T = TypeVar("T")
10U = TypeVar("U")
11
12

[docs]
13def map_(
14    func: Callable[[T], U], /, *, lazy: bool = True
15) -> Operation[Iterable[T], Iterable[U]]:
16    def _execute(iterable: Iterable[T]) -> Iterable[U]:
17        if lazy:
18            return map(func, iterable)
19        return [func(a) for a in iterable]
20
21    return _execute

22
23

[docs]
24def filter_(
25    func: Callable[[T], bool], /, *, lazy: bool = True
26) -> Operation[Iterable[T], Iterable[T]]:
27    def _execute(iterable: Iterable[T]) -> Iterable[T]:
28        if lazy:
29            return filter(func, iterable)
30        return [a for a in iterable if func(a)]
31
32    return _execute

33
34

[docs]
35def filter_none(
36    *, lazy: bool = True
37) -> Operation[Iterable[Union[T, None]], Iterable[T]]:
38    def _execute(iterable: Iterable[Union[T, None]]) -> Iterable[T]:
39        if lazy:
40            return (a for a in iterable if a is not None)
41        return [a for a in iterable if a is not None]
42
43    return _execute

44
45

[docs]
46def reduce(initial: T, func: Callable[[T, U], T]) -> Operation[Iterable[U], T]:
47    def _execute(iterable: Iterable[U]) -> T:
48        return functools.reduce(func, iterable, initial)
49
50    return _execute

51
52

[docs]
53def collect() -> Operation[Iterable[T], list[T]]:
54    return reduce([], lambda a, b: a + [b])