Skip to content

Core API

eaterate.eater(it)

Creates an Eaterator object from either an iterable, an iterator, or an eaterator.

  • Iterable: something that can create a Iterator from __iter__.
  • Iterator: something that can be iterated with __next__.
  • Eaterator: iterators with additional features.
Example
r = range(100)
eat: Eaterator = eater(r)

Parameters:

Name Type Description Default
it Iterable | Iterator | Eaterator

The iterator or iterable.

required

Returns:

Type Description
Eaterator[T]

An Eaterator (iterator) object.

Raises:

Type Description
TypeError

The provided object is not an iterable, an iterator, or an Eaterator object.

Source code in eaterate/core.py
def eater(it: "AutoIt[T]") -> "Eaterator[T]":
    """Creates an `Eaterator` object from either an iterable, an iterator, or an eaterator.

    - Iterable: something that can create a `Iterator` from `__iter__`.
    - Iterator: something that can be iterated with `__next__`.
    - Eaterator: iterators with additional features.

    Example:
        ```python
        r = range(100)
        eat: Eaterator = eater(r)
        ```

    Args:
        it (Iterable | Iterator | Eaterator): The iterator or iterable.

    Returns:
        An `Eaterator` (iterator) object.

    Raises:
        TypeError: The provided object is not an iterable, an iterator, or an Eaterator object.
    """
    if hasattr(it, "__next__"):
        return BuiltinItEaterator(it)  # type: ignore
    elif isinstance(it, Eaterator):
        return it
    elif hasattr(it, "__iter__"):
        return BuiltinItEaterator(it.__iter__())
    else:
        raise TypeError(
            f"expected either an iterable, an iterator, or an Eaterator object, got: {type(it)!r}"
        )

eaterate.erange(a, b, c=None)

erange(a: EllipsisType, b: int, c: None = None) -> ERange
erange(a: int, b: EllipsisType, c: None = None) -> Eaterator[int]
erange(a: int, b: EllipsisType, c: int) -> ERange

Creates a range.

Example

To create a range that starts from 0, stops at 3:

erange(0, ..., 3)

# to include 3:
erange(0, ..., 3).inclusive()

Alternatively:

erange(..., 3)

# to include 3:
erange(..., 3).inclusive()

To create a range that starts from 0, yet never stops (infinite):

# inclusive() is not available
erange(0, ...)
Source code in eaterate/range.py
def erange(
    a: Union[EllipsisType, int],
    b: Union[EllipsisType, int],
    c: Optional[Union[EllipsisType, int]] = None,
) -> Union[ERange, Eaterator[int]]:
    """Creates a range.

    Example:
        To create a range that **starts from `0`, stops at `3`**:

        ```python
        erange(0, ..., 3)

        # to include 3:
        erange(0, ..., 3).inclusive()
        ```

        Alternatively:

        ```python
        erange(..., 3)

        # to include 3:
        erange(..., 3).inclusive()
        ```

        To create a range that **starts from `0`, yet never stops (infinite)**:

        ```python
        # inclusive() is not available
        erange(0, ...)
        ```
    """
    if isinstance(a, int):
        if isinstance(c, int):
            return ERange(a, c)
        return ERange(a, float("inf"))  # type: ignore

    else:
        return ERange(0, b)  # type: ignore

eaterate.Eaterator

Bases: Generic[T]

Iterator with additional features.

Supports for loops.

Example
eat = eater([1, 2, 3]).chain([4, 5, 6])
eat.collect(list)  # [1, 2, 3, 4, 5, 6]

You can also use a for loop:

eat = eater([1, 2, 3]).chain([4, 5, 6])

for i in eat:
    print(i)

Source code in eaterate/core.py
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
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
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
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
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
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
322
323
324
325
326
327
328
329
330
331
332
333
334
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
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
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
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
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
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
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
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
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
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
class Eaterator(Generic[T]):
    """Iterator with additional features.

    Supports `for` loops.

    Example:
        ```python
        eat = eater([1, 2, 3]).chain([4, 5, 6])
        eat.collect(list)  # [1, 2, 3, 4, 5, 6]
        ```

        You can also use a `for` loop:
        ```python
        eat = eater([1, 2, 3]).chain([4, 5, 6])

        for i in eat:
            print(i)
        ```
    """

    def next(self) -> Option[T]:
        """**Required method**.

        Iterates to the next item.

        On the user's interface, it can also be interpreted as 'the first item' if
        at the start of the iterator.

        Example:
            ```python
            class MyEaterator(Eaterator[int]):
                def next(self) -> Option[int]:
                    if exhausted:
                        # the iterator stops when Option.none() is present
                        return Option.none()
                    else:
                        # this is the actual value you'd like to yield
                        return Option.some(1)

            ```

        Returns:
            `Option.none()` if the iteration should stop.
        """
        raise NotImplementedError

    def map(self, fn: Callable[[T], K], /) -> "MapEaterator[T, K]":
        """Map the elements of this iterator.

        Args:
            fn: Function to transform each element.
        """
        return MapEaterator(self, fn)

    def all(self, fn: Callable[[T], bool], /) -> bool:
        """Tests if every element of the iterator matches a predicate.

        Equivalents to Python's `all()`.
        """
        while True:
            x = self.next().map(fn)
            if x.is_none():
                return True

            if not x._unwrap():
                return False

    def any(self, fn: Callable[[T], bool], /) -> bool:
        """Tests if an element of the iterator matches a predicate.

        Equivalents to Python's `any()`.
        """
        while True:
            x = self.next().map(fn)
            if x.is_none():
                return False

            if x._unwrap():
                return True

    def find(self, fn: Callable[[T], bool], /) -> Option[T]:
        """Searches for an element of the iterator that satisfies a predicate.

        Example:
            ```python
            eat = eater([1, 2, 3]).find(lambda x: x % 2 == 0)
            print(eat)  # Some(2)
            ```

        Returns:
            Option[T]: An `Option` object, which is **NOT** `typing.Optional[T]`.
        """
        while True:
            x = self.next()
            if x.is_none():
                return Option.none()

            if fn(x._unwrap()):
                return x

    def count(self) -> int:
        """Consumes the iterator, counting the number of iterations and returning it.

        Example:
            ```python
            eat = eater(range(10)).count()
            print(eat)  # 10
            ```
        """
        x = 0
        while True:
            if self.next().is_none():
                break
            x += 1
        return x

    def last(self) -> Option[T]:
        """Consumes the iterator, returning the last element.

        This method will evaluate the iterator until it returns the Option.none().
        """
        x = Option.none()
        while True:
            t = self.next()
            if t.is_none():
                break
            x = t
        return x

    def nth(self, n: int, /) -> Option[T]:
        """Returns the `n`-th element of the iterator."""
        assert n >= 0, "requires: n >= 0"

        while True:
            x = self.next()

            if n == 0:
                return x
            elif x.is_none():
                return Option.none()

            n -= 1

    def step_by(self, step: int, /) -> "StepByEaterator[T]":
        """Creates an iterator starting at the same point, but stepping by `step` at each iteration.

        This implementation ensures no number greater than `step + 1` is used.

        Example:
            ```python
            eat = eater([0, 1, 2, 3, 4, 5]).step_by(2)

            print(eat.next())  # Some(0)
            print(eat.next())  # Some(2)
            print(eat.next())  # Some(4)
            print(eat.next())  # Option.none()
            ```
        """
        if step == 1:
            return self  # type: ignore
        return StepByEaterator(self, step)

    def chain(self, *eats: "AutoIt[T]") -> "ChainEaterator[T]":
        """Chain multiple iterators into one.

        Args:
            *eats (`AutoIt[T]`): Other iterators.
        """
        e = ChainEaterator(self, eater(eats[0]))
        for itm in eats[1:]:
            e = ChainEaterator(e, eater(itm))
        return e

    def zip(self, eat: "AutoIt[K]", /) -> "ZipEaterator[T, K]":
        """'Zips up' two iterators into a single iterator of pairs.

        This returns a new iterator that will iterate over two other iterators, returning a tuple
        where the first element comes from the first iterator, and the second element comes from the second iterator.

        Stops when either one of them has stopped.

        This behaves like Python's built-in `zip()`, except only accepting one iterator only.

        Examples:

        (1) You can simply pass in two iterators.

        ```python
        eat = eater([0, 1, 2]).zip([1, 2, 3])

        print(eat.next())  # Some((0, 1))
        print(eat.next())  # Some((1, 2))
        print(eat.next())  # Some((2, 3))
        print(eat.next())  # Option.none()
        ```

        (2) Sometimes their lengths don't match. It stops whenever one of the two iterators stops.

        ```python
        eat = eater([0, 1, 2]).zip([1, 2, 3, 4, 5])

        print(eat.next())  # Some((0, 1))
        print(eat.next())  # Some((1, 2))
        print(eat.next())  # Some((2, 3))
        print(eat.next())  # Option.none()
        ```

        (3) When extracting more than two zipped iterators, beware of the `(tuple)` syntax.

        ```python
        eat = eater([0, 1, 2]).zip([2, 3, 4]).zip([4, 5, 6])

        for (a, b), c in it:
            print(a, b, c)
        ```

        Args:
            eat: The other iterator.
        """
        return ZipEaterator(self, eater(eat))

    def intersperse(self, sep: T, /) -> "IntersperseEaterator[T]":
        """Creates a new iterator which places a reference of `sep` (separator) between adjacent elements of the original iterator.

        Example:
            ```python
            eat = eater([0, 1, 2]).intersperse(10)

            print(eat.next())  # Some(0)
            print(eat.next())  # Some(10)
            print(eat.next())  # Some(1)
            print(eat.next())  # Some(10)
            print(eat.next())  # Some(2)
            print(eat.next())  # Option.none()
            ```

        Args:
            sep: The separator.
        """
        return IntersperseEaterator(self, sep)

    def for_each(self, fn: Callable[[T], Any], /) -> None:
        """Calls a function on each element of this iterator.

        To make your code Pythonic, it's recommended to just use a `for` loop.

        Example:
            ```python
            eat = eater([0, 1, 2])
            eat.for_each(lambda x: print(x))

            # Output:
            # 0
            # 1
            # 2
            ```

        Args:
            fn: The function. Takes one parameter: an element.
        """
        while True:
            x = self.next()
            if x.is_none():
                break
            fn(x._unwrap())

    def try_for_each(
        self, fn: Callable[[T], Any], _errhint: type[E] = Exception, /
    ) -> Union[E, None]:  # not to be confused with Option
        """Calls a falliable function on each element of this iterator.

        Stops when one iteration has an error (exception) occurred.

        Example:
            Let's assume you have a function defined for `try_for_each` that may fail, as well as
            an iterator. You'll notice that `try_for_each` gracefully catches the error, and returns it.
            ```python
            def nah(x: int):
                raise RuntimeError("hell nawh!")

            # the iterator
            eat = eater([1, 2, 3])

            err = eat.try_for_each(nah)
            if err is not None:
                print(err)  # hell nawh!
            else:
                print('ok')
            ```

            If needed, you can also provide the type checker with exception hints.
            If provided, only that exception will be caught.

            ```python
            eat.try_for_each(nah, RuntimeError)
            ```

        Args:
            fn (Callable): The function. Takes one parameter: an element.
            _errhint (Exception, optional): Type hint that specifies what error may occur or be caught.
        """
        while True:
            x = self.next()
            if x.is_none():
                break
            try:
                fn(x._unwrap())
            except _errhint as err:
                return err

    def filter(self, fn: Callable[[T], bool], /) -> "FilterEaterator[T]":
        """Creates an iterator which uses a function to determine if an element should be yielded.

        Example:
            ```python
            eat = eater(range(5)).filter(lambda i: i % 2 == 0)

            print(eat.next())  # Some(0)
            print(eat.next())  # Some(2)
            print(eat.next())  # Some(4)
            print(eat.next())  # Option.none()
            ```

        Args:
            fn: The function. Takes one parameter: an element.
        """
        return FilterEaterator(self, fn)

    def enumerate(self) -> "EnumerateEaterator[T]":
        """Creates an iterator which gives the current iteration count as well as the value.

        The iterator yields pairs `(i, val)`.

        - `i`: the current index of iteration.
        - `val`: the value returned by the original iterator.

        Example:
            ```python
            eat = eater("hi!").enumerate()

            print(eat.next())  # Some((0, "h"))
            print(eat.next())  # Some((1, "i"))
            print(eat.next())  # Some((2, "!"))
            print(eat.next())  # Option.none()
            ```
        """
        return EnumerateEaterator(self)

    def peeked(self) -> "PeekedEaterator":
        """Creates an iterator that gives the current value and the next one, allowing you to peek into the next data.

        For each element, you get `(current, peeked)`, where:

        - current: the current value.
        - peeked: an `Option`, which could be `Option.none()` if no data is ahead.

        If you'd like to receive more than one element at a time, see :meth:`windows`, which features a more complex implementation.

        Example:
            ```python
            eat = eater("hi!").peeked()

            print(eat.next())  # Some(("h", Some("i")))
            print(eat.next())  # Some(("i", Some("!")))
            print(eat.next())  # Some(("!", Option.none()))
            print(eat.next())  # Option.none()
            ```
        """
        return PeekedEaterator(self)

    def skip(self, n: int, /) -> "SkipEaterator[T]":
        """Skip the first `n` elements.

        Args:
            n: Number of elements.
        """
        return SkipEaterator(self, n)

    def take(self, n: int, /) -> "TakeEaterator[T]":
        """Creates an iterator that only yields the first `n` elements.

        May be fewer than the requested amount.

        Args:
            n: Number of elements.
        """
        return TakeEaterator(self, n)

    @overload
    def collect(self, dst: type[list[T]], /) -> list[T]: ...

    @overload
    def collect(self, dst: type[list[T]] = list, /) -> list[T]: ...

    @overload
    def collect(self, dst: type[deque[T]], /) -> deque[T]: ...

    @overload
    def collect(self, dst: type[dict[int, T]], /) -> dict[int, T]: ...

    @overload
    def collect(self, dst: type[str], /) -> str: ...

    @overload
    def collect(self, dst: type[set], /) -> set[T]: ...

    def collect(
        self, dst: type[Union[list[T], deque[T], dict[int, T], str, set]] = list, /
    ) -> Union[list[T], deque[T], dict[int, T], str, set]:
        """Collect items by iterating over all items. Defaults to `list`.

        You can choose one of:

        - `list[T]`: collects to a list. **Default behavior**.
        - `deque[T]`: collects to a deque. (See `collect_deque()` for more options)
        - `dict[int, T]`: collects to a dictionary, with index keys.
        - `str`: collects to a string.
        - `set`: collects to a set.

        Example:
            ```python
            eat.collect(list)
            eat.collect(deque)
            eat.collect(dict)
            eat.collect(str)
            eat.collect(set)
            ```

            You can add additional annotations, if needed:
            ```python
            # eaterate won't read 'int', it only recognizes 'list'
            # you need to ensure the type yourself, both in type
            # checking and runtime
            eat.collect(list[int])
            ```
        """
        # if no origin, possibly the user didn't use any typevar
        origin = typing.get_origin(dst) or dst

        if origin is list:
            return self.collect_list()
        elif origin is deque:
            return self.collect_deque()
        elif origin is str:
            return self.collect_str()
        elif origin is dict:
            return self.collect_enumerated_dict()
        elif origin is set:
            return self.collect_set()
        else:
            raise NotImplementedError(f"unknown collector: {origin!r} (from: {dst!r})")

    def collect_list(self) -> list[T]:
        """Collect items of this iterator to a `dict`."""
        arr = []
        while True:
            x = self.next()
            if x.is_none():
                break
            arr.append(x._unwrap())
        return arr

    def collect_deque(self, *, reverse: bool = False) -> deque[T]:
        """Collect items of this iterator to a `deque`.

        Args:
            reverse (bool, optional): Whether to reverse the order.
                Defaults to `False`.
        """
        d = deque()
        while True:
            x = self.next()
            if x.is_none():
                break
            if reverse:
                d.appendleft(x._unwrap())
            else:
                d.append(x._unwrap())
        return d

    def collect_enumerated_dict(self) -> dict[int, T]:
        """Collect items of this iterator to a `dict`, with index numbers as the key.

        In other words, you may get a dictionary like this:
        ```python
        {
            0: "h",
            1: "i",
            2: "!",
        }
        ```

        ...which is zero-indexed.

        To keep it simple, this function does not use `EnumerateEaterator` iterator.

        You can also use the `collect(dict)` instead.
        """
        d = dict()
        i = 0
        while True:
            x = self.next()
            if x.is_none():
                break
            d[i] = x._unwrap()
            i += 1
        return d

    def collect_str(self) -> str:
        """Collect items of this iterator to a `str`.

        Example:
            ```python
            eat = eater(["m", "o", "n", "e", "y"])
            eat.collect_str()  # money
            ```
        """
        s = ""
        while True:
            x = self.next()
            if x.is_none():
                break
            s += str(x._unwrap())
        return s

    def collect_set(self) -> "set[T]":
        """Collects items of this iterator to a `set`, which ensures there are no repeated items.

        Example:
            ```python
            res = eater([0, 0, 1, 2]).collect_set()
            print(res)  # {0, 1, 2}
            ```
        """
        return set(self)

    def flatten(self) -> "FlattenEaterator[T]":
        """Creates an iterator that flattens nested structure.

        This is useful when you have *an iterator of iterators* or *an iterator of elements* that can be turned into iterators,
        and you'd like to flatten them to one layer only.

        **Important**: **requires each element to satisfy `Iterable[K] | Iterator[K] | Eaterator[K]`** (`AutoIt`).

        Example:
            ```python
            eat = (
                eater([
                    ["hello", "world"],
                    ["multi", "layer"]
                ])
                .flatten()
            )

            eat.next()  # Some("hello")
            eat.next()  # Some("world")
            eat.next()  # Some("multi")
            eat.next()  # Some("layer")
            eat.next()  # Option.none()
            ```
        """
        return FlattenEaterator(self)  # type: ignore

    def fold(self, init: K, fn: Callable[[K, T], K], /) -> K:
        """Folds every element into an accumulator by applying an operation, returning the final result.

        Example:
            ```python
            res = (
                eater([1, 2, 3])
                .fold("0", lambda acc, x: f"({acc} + {x})")
            )

            print(res)  # (((0 + 1) + 2) + 3)
            ```

        Args:
            init: The initial value.
            fn: The accumlator function.
        """
        while True:
            x = self.next()
            if x.is_none():
                break
            init = fn(init, x._unwrap())

        return init

    def windows(self, size: int) -> "WindowsEaterator[T]":
        """Creates an iterator over overlapping subslices of length `size`.

        Example:
            ```python
            eat = eater([1, 2, 3, 4]).windows(2)

            print(eat.next())  # Some([1, 2])
            print(eat.next())  # Some([2, 3])
            print(eat.next())  # Some([3, 4])
            print(eat.next())  # Option.none()
            ```

            When `size` is greater than the *actual size* of the original iterator, this
            immediately stops.

            ```python
            eat = eater([1, 2, 3]).windows(5)
            print(eat.next())  # Option.none()
            ```
        """
        return WindowsEaterator(self, size)

    def __iter__(self) -> Iterator[T]:
        return self

    # IMPORTANT:
    # Somehow Python executes __len__ when unpacking,
    # which is REALLY bad, since using count() consumes
    # the iterator. Therefore this feature is deprecated
    # for good.

    # def __len__(self) -> int:
    #     return self.count()

    def __next__(self) -> T:
        x = self.next()
        if x.is_some():
            return x._unwrap()
        else:
            raise StopIteration

    @overload
    def __getitem__(self, key: slice) -> "Eaterator[T]": ...

    @overload
    def __getitem__(self, key: int) -> T: ...

    def __getitem__(self, key: Union[slice, int]) -> Union[T, "Eaterator[T]"]:
        if isinstance(key, int):
            x = self.nth(key)
            if x.is_none():
                raise IndexError(f"index out of range (requested: {key})")
            else:
                return x._unwrap()
        else:
            start = key.start or 0
            stop = key.stop or 1
            step = key.step or 1

            if start < 0 or stop < 0 or step < 0:
                raise ValueError("any of slice(start, stop, step) cannot be negative")

            return self.skip(start).take(stop - start).step_by(step)

    def __repr__(self) -> str:
        return "Eaterator(...)"

all(fn)

Tests if every element of the iterator matches a predicate.

Equivalents to Python's all().

Source code in eaterate/core.py
def all(self, fn: Callable[[T], bool], /) -> bool:
    """Tests if every element of the iterator matches a predicate.

    Equivalents to Python's `all()`.
    """
    while True:
        x = self.next().map(fn)
        if x.is_none():
            return True

        if not x._unwrap():
            return False

any(fn)

Tests if an element of the iterator matches a predicate.

Equivalents to Python's any().

Source code in eaterate/core.py
def any(self, fn: Callable[[T], bool], /) -> bool:
    """Tests if an element of the iterator matches a predicate.

    Equivalents to Python's `any()`.
    """
    while True:
        x = self.next().map(fn)
        if x.is_none():
            return False

        if x._unwrap():
            return True

chain(*eats)

Chain multiple iterators into one.

Parameters:

Name Type Description Default
*eats `AutoIt[T]`

Other iterators.

()
Source code in eaterate/core.py
def chain(self, *eats: "AutoIt[T]") -> "ChainEaterator[T]":
    """Chain multiple iterators into one.

    Args:
        *eats (`AutoIt[T]`): Other iterators.
    """
    e = ChainEaterator(self, eater(eats[0]))
    for itm in eats[1:]:
        e = ChainEaterator(e, eater(itm))
    return e

collect(dst=list)

collect(dst: type[list[T]]) -> list[T]
collect(dst: type[list[T]] = list) -> list[T]
collect(dst: type[deque[T]]) -> deque[T]
collect(dst: type[dict[int, T]]) -> dict[int, T]
collect(dst: type[str]) -> str
collect(dst: type[set]) -> set[T]

Collect items by iterating over all items. Defaults to list.

You can choose one of:

  • list[T]: collects to a list. Default behavior.
  • deque[T]: collects to a deque. (See collect_deque() for more options)
  • dict[int, T]: collects to a dictionary, with index keys.
  • str: collects to a string.
  • set: collects to a set.
Example
eat.collect(list)
eat.collect(deque)
eat.collect(dict)
eat.collect(str)
eat.collect(set)

You can add additional annotations, if needed:

# eaterate won't read 'int', it only recognizes 'list'
# you need to ensure the type yourself, both in type
# checking and runtime
eat.collect(list[int])

Source code in eaterate/core.py
def collect(
    self, dst: type[Union[list[T], deque[T], dict[int, T], str, set]] = list, /
) -> Union[list[T], deque[T], dict[int, T], str, set]:
    """Collect items by iterating over all items. Defaults to `list`.

    You can choose one of:

    - `list[T]`: collects to a list. **Default behavior**.
    - `deque[T]`: collects to a deque. (See `collect_deque()` for more options)
    - `dict[int, T]`: collects to a dictionary, with index keys.
    - `str`: collects to a string.
    - `set`: collects to a set.

    Example:
        ```python
        eat.collect(list)
        eat.collect(deque)
        eat.collect(dict)
        eat.collect(str)
        eat.collect(set)
        ```

        You can add additional annotations, if needed:
        ```python
        # eaterate won't read 'int', it only recognizes 'list'
        # you need to ensure the type yourself, both in type
        # checking and runtime
        eat.collect(list[int])
        ```
    """
    # if no origin, possibly the user didn't use any typevar
    origin = typing.get_origin(dst) or dst

    if origin is list:
        return self.collect_list()
    elif origin is deque:
        return self.collect_deque()
    elif origin is str:
        return self.collect_str()
    elif origin is dict:
        return self.collect_enumerated_dict()
    elif origin is set:
        return self.collect_set()
    else:
        raise NotImplementedError(f"unknown collector: {origin!r} (from: {dst!r})")

collect_deque(*, reverse=False)

Collect items of this iterator to a deque.

Parameters:

Name Type Description Default
reverse bool

Whether to reverse the order. Defaults to False.

False
Source code in eaterate/core.py
def collect_deque(self, *, reverse: bool = False) -> deque[T]:
    """Collect items of this iterator to a `deque`.

    Args:
        reverse (bool, optional): Whether to reverse the order.
            Defaults to `False`.
    """
    d = deque()
    while True:
        x = self.next()
        if x.is_none():
            break
        if reverse:
            d.appendleft(x._unwrap())
        else:
            d.append(x._unwrap())
    return d

collect_enumerated_dict()

Collect items of this iterator to a dict, with index numbers as the key.

In other words, you may get a dictionary like this:

{
    0: "h",
    1: "i",
    2: "!",
}

...which is zero-indexed.

To keep it simple, this function does not use EnumerateEaterator iterator.

You can also use the collect(dict) instead.

Source code in eaterate/core.py
def collect_enumerated_dict(self) -> dict[int, T]:
    """Collect items of this iterator to a `dict`, with index numbers as the key.

    In other words, you may get a dictionary like this:
    ```python
    {
        0: "h",
        1: "i",
        2: "!",
    }
    ```

    ...which is zero-indexed.

    To keep it simple, this function does not use `EnumerateEaterator` iterator.

    You can also use the `collect(dict)` instead.
    """
    d = dict()
    i = 0
    while True:
        x = self.next()
        if x.is_none():
            break
        d[i] = x._unwrap()
        i += 1
    return d

collect_list()

Collect items of this iterator to a dict.

Source code in eaterate/core.py
def collect_list(self) -> list[T]:
    """Collect items of this iterator to a `dict`."""
    arr = []
    while True:
        x = self.next()
        if x.is_none():
            break
        arr.append(x._unwrap())
    return arr

collect_set()

Collects items of this iterator to a set, which ensures there are no repeated items.

Example
res = eater([0, 0, 1, 2]).collect_set()
print(res)  # {0, 1, 2}
Source code in eaterate/core.py
def collect_set(self) -> "set[T]":
    """Collects items of this iterator to a `set`, which ensures there are no repeated items.

    Example:
        ```python
        res = eater([0, 0, 1, 2]).collect_set()
        print(res)  # {0, 1, 2}
        ```
    """
    return set(self)

collect_str()

Collect items of this iterator to a str.

Example
eat = eater(["m", "o", "n", "e", "y"])
eat.collect_str()  # money
Source code in eaterate/core.py
def collect_str(self) -> str:
    """Collect items of this iterator to a `str`.

    Example:
        ```python
        eat = eater(["m", "o", "n", "e", "y"])
        eat.collect_str()  # money
        ```
    """
    s = ""
    while True:
        x = self.next()
        if x.is_none():
            break
        s += str(x._unwrap())
    return s

count()

Consumes the iterator, counting the number of iterations and returning it.

Example
eat = eater(range(10)).count()
print(eat)  # 10
Source code in eaterate/core.py
def count(self) -> int:
    """Consumes the iterator, counting the number of iterations and returning it.

    Example:
        ```python
        eat = eater(range(10)).count()
        print(eat)  # 10
        ```
    """
    x = 0
    while True:
        if self.next().is_none():
            break
        x += 1
    return x

enumerate()

Creates an iterator which gives the current iteration count as well as the value.

The iterator yields pairs (i, val).

  • i: the current index of iteration.
  • val: the value returned by the original iterator.
Example
eat = eater("hi!").enumerate()

print(eat.next())  # Some((0, "h"))
print(eat.next())  # Some((1, "i"))
print(eat.next())  # Some((2, "!"))
print(eat.next())  # Option.none()
Source code in eaterate/core.py
def enumerate(self) -> "EnumerateEaterator[T]":
    """Creates an iterator which gives the current iteration count as well as the value.

    The iterator yields pairs `(i, val)`.

    - `i`: the current index of iteration.
    - `val`: the value returned by the original iterator.

    Example:
        ```python
        eat = eater("hi!").enumerate()

        print(eat.next())  # Some((0, "h"))
        print(eat.next())  # Some((1, "i"))
        print(eat.next())  # Some((2, "!"))
        print(eat.next())  # Option.none()
        ```
    """
    return EnumerateEaterator(self)

filter(fn)

Creates an iterator which uses a function to determine if an element should be yielded.

Example
eat = eater(range(5)).filter(lambda i: i % 2 == 0)

print(eat.next())  # Some(0)
print(eat.next())  # Some(2)
print(eat.next())  # Some(4)
print(eat.next())  # Option.none()

Parameters:

Name Type Description Default
fn Callable[[T], bool]

The function. Takes one parameter: an element.

required
Source code in eaterate/core.py
def filter(self, fn: Callable[[T], bool], /) -> "FilterEaterator[T]":
    """Creates an iterator which uses a function to determine if an element should be yielded.

    Example:
        ```python
        eat = eater(range(5)).filter(lambda i: i % 2 == 0)

        print(eat.next())  # Some(0)
        print(eat.next())  # Some(2)
        print(eat.next())  # Some(4)
        print(eat.next())  # Option.none()
        ```

    Args:
        fn: The function. Takes one parameter: an element.
    """
    return FilterEaterator(self, fn)

find(fn)

Searches for an element of the iterator that satisfies a predicate.

Example
eat = eater([1, 2, 3]).find(lambda x: x % 2 == 0)
print(eat)  # Some(2)

Returns:

Type Description
Option[T]

Option[T]: An Option object, which is NOT typing.Optional[T].

Source code in eaterate/core.py
def find(self, fn: Callable[[T], bool], /) -> Option[T]:
    """Searches for an element of the iterator that satisfies a predicate.

    Example:
        ```python
        eat = eater([1, 2, 3]).find(lambda x: x % 2 == 0)
        print(eat)  # Some(2)
        ```

    Returns:
        Option[T]: An `Option` object, which is **NOT** `typing.Optional[T]`.
    """
    while True:
        x = self.next()
        if x.is_none():
            return Option.none()

        if fn(x._unwrap()):
            return x

flatten()

Creates an iterator that flattens nested structure.

This is useful when you have an iterator of iterators or an iterator of elements that can be turned into iterators, and you'd like to flatten them to one layer only.

Important: requires each element to satisfy Iterable[K] | Iterator[K] | Eaterator[K] (AutoIt).

Example
eat = (
    eater([
        ["hello", "world"],
        ["multi", "layer"]
    ])
    .flatten()
)

eat.next()  # Some("hello")
eat.next()  # Some("world")
eat.next()  # Some("multi")
eat.next()  # Some("layer")
eat.next()  # Option.none()
Source code in eaterate/core.py
def flatten(self) -> "FlattenEaterator[T]":
    """Creates an iterator that flattens nested structure.

    This is useful when you have *an iterator of iterators* or *an iterator of elements* that can be turned into iterators,
    and you'd like to flatten them to one layer only.

    **Important**: **requires each element to satisfy `Iterable[K] | Iterator[K] | Eaterator[K]`** (`AutoIt`).

    Example:
        ```python
        eat = (
            eater([
                ["hello", "world"],
                ["multi", "layer"]
            ])
            .flatten()
        )

        eat.next()  # Some("hello")
        eat.next()  # Some("world")
        eat.next()  # Some("multi")
        eat.next()  # Some("layer")
        eat.next()  # Option.none()
        ```
    """
    return FlattenEaterator(self)  # type: ignore

fold(init, fn)

Folds every element into an accumulator by applying an operation, returning the final result.

Example
res = (
    eater([1, 2, 3])
    .fold("0", lambda acc, x: f"({acc} + {x})")
)

print(res)  # (((0 + 1) + 2) + 3)

Parameters:

Name Type Description Default
init K

The initial value.

required
fn Callable[[K, T], K]

The accumlator function.

required
Source code in eaterate/core.py
def fold(self, init: K, fn: Callable[[K, T], K], /) -> K:
    """Folds every element into an accumulator by applying an operation, returning the final result.

    Example:
        ```python
        res = (
            eater([1, 2, 3])
            .fold("0", lambda acc, x: f"({acc} + {x})")
        )

        print(res)  # (((0 + 1) + 2) + 3)
        ```

    Args:
        init: The initial value.
        fn: The accumlator function.
    """
    while True:
        x = self.next()
        if x.is_none():
            break
        init = fn(init, x._unwrap())

    return init

for_each(fn)

Calls a function on each element of this iterator.

To make your code Pythonic, it's recommended to just use a for loop.

Example
eat = eater([0, 1, 2])
eat.for_each(lambda x: print(x))

# Output:
# 0
# 1
# 2

Parameters:

Name Type Description Default
fn Callable[[T], Any]

The function. Takes one parameter: an element.

required
Source code in eaterate/core.py
def for_each(self, fn: Callable[[T], Any], /) -> None:
    """Calls a function on each element of this iterator.

    To make your code Pythonic, it's recommended to just use a `for` loop.

    Example:
        ```python
        eat = eater([0, 1, 2])
        eat.for_each(lambda x: print(x))

        # Output:
        # 0
        # 1
        # 2
        ```

    Args:
        fn: The function. Takes one parameter: an element.
    """
    while True:
        x = self.next()
        if x.is_none():
            break
        fn(x._unwrap())

intersperse(sep)

Creates a new iterator which places a reference of sep (separator) between adjacent elements of the original iterator.

Example
eat = eater([0, 1, 2]).intersperse(10)

print(eat.next())  # Some(0)
print(eat.next())  # Some(10)
print(eat.next())  # Some(1)
print(eat.next())  # Some(10)
print(eat.next())  # Some(2)
print(eat.next())  # Option.none()

Parameters:

Name Type Description Default
sep T

The separator.

required
Source code in eaterate/core.py
def intersperse(self, sep: T, /) -> "IntersperseEaterator[T]":
    """Creates a new iterator which places a reference of `sep` (separator) between adjacent elements of the original iterator.

    Example:
        ```python
        eat = eater([0, 1, 2]).intersperse(10)

        print(eat.next())  # Some(0)
        print(eat.next())  # Some(10)
        print(eat.next())  # Some(1)
        print(eat.next())  # Some(10)
        print(eat.next())  # Some(2)
        print(eat.next())  # Option.none()
        ```

    Args:
        sep: The separator.
    """
    return IntersperseEaterator(self, sep)

last()

Consumes the iterator, returning the last element.

This method will evaluate the iterator until it returns the Option.none().

Source code in eaterate/core.py
def last(self) -> Option[T]:
    """Consumes the iterator, returning the last element.

    This method will evaluate the iterator until it returns the Option.none().
    """
    x = Option.none()
    while True:
        t = self.next()
        if t.is_none():
            break
        x = t
    return x

map(fn)

Map the elements of this iterator.

Parameters:

Name Type Description Default
fn Callable[[T], K]

Function to transform each element.

required
Source code in eaterate/core.py
def map(self, fn: Callable[[T], K], /) -> "MapEaterator[T, K]":
    """Map the elements of this iterator.

    Args:
        fn: Function to transform each element.
    """
    return MapEaterator(self, fn)

next()

Required method.

Iterates to the next item.

On the user's interface, it can also be interpreted as 'the first item' if at the start of the iterator.

Example
class MyEaterator(Eaterator[int]):
    def next(self) -> Option[int]:
        if exhausted:
            # the iterator stops when Option.none() is present
            return Option.none()
        else:
            # this is the actual value you'd like to yield
            return Option.some(1)

Returns:

Type Description
Option[T]

Option.none() if the iteration should stop.

Source code in eaterate/core.py
def next(self) -> Option[T]:
    """**Required method**.

    Iterates to the next item.

    On the user's interface, it can also be interpreted as 'the first item' if
    at the start of the iterator.

    Example:
        ```python
        class MyEaterator(Eaterator[int]):
            def next(self) -> Option[int]:
                if exhausted:
                    # the iterator stops when Option.none() is present
                    return Option.none()
                else:
                    # this is the actual value you'd like to yield
                    return Option.some(1)

        ```

    Returns:
        `Option.none()` if the iteration should stop.
    """
    raise NotImplementedError

nth(n)

Returns the n-th element of the iterator.

Source code in eaterate/core.py
def nth(self, n: int, /) -> Option[T]:
    """Returns the `n`-th element of the iterator."""
    assert n >= 0, "requires: n >= 0"

    while True:
        x = self.next()

        if n == 0:
            return x
        elif x.is_none():
            return Option.none()

        n -= 1

peeked()

Creates an iterator that gives the current value and the next one, allowing you to peek into the next data.

For each element, you get (current, peeked), where:

  • current: the current value.
  • peeked: an Option, which could be Option.none() if no data is ahead.

If you'd like to receive more than one element at a time, see :meth:windows, which features a more complex implementation.

Example
eat = eater("hi!").peeked()

print(eat.next())  # Some(("h", Some("i")))
print(eat.next())  # Some(("i", Some("!")))
print(eat.next())  # Some(("!", Option.none()))
print(eat.next())  # Option.none()
Source code in eaterate/core.py
def peeked(self) -> "PeekedEaterator":
    """Creates an iterator that gives the current value and the next one, allowing you to peek into the next data.

    For each element, you get `(current, peeked)`, where:

    - current: the current value.
    - peeked: an `Option`, which could be `Option.none()` if no data is ahead.

    If you'd like to receive more than one element at a time, see :meth:`windows`, which features a more complex implementation.

    Example:
        ```python
        eat = eater("hi!").peeked()

        print(eat.next())  # Some(("h", Some("i")))
        print(eat.next())  # Some(("i", Some("!")))
        print(eat.next())  # Some(("!", Option.none()))
        print(eat.next())  # Option.none()
        ```
    """
    return PeekedEaterator(self)

skip(n)

Skip the first n elements.

Parameters:

Name Type Description Default
n int

Number of elements.

required
Source code in eaterate/core.py
def skip(self, n: int, /) -> "SkipEaterator[T]":
    """Skip the first `n` elements.

    Args:
        n: Number of elements.
    """
    return SkipEaterator(self, n)

step_by(step)

Creates an iterator starting at the same point, but stepping by step at each iteration.

This implementation ensures no number greater than step + 1 is used.

Example
eat = eater([0, 1, 2, 3, 4, 5]).step_by(2)

print(eat.next())  # Some(0)
print(eat.next())  # Some(2)
print(eat.next())  # Some(4)
print(eat.next())  # Option.none()
Source code in eaterate/core.py
def step_by(self, step: int, /) -> "StepByEaterator[T]":
    """Creates an iterator starting at the same point, but stepping by `step` at each iteration.

    This implementation ensures no number greater than `step + 1` is used.

    Example:
        ```python
        eat = eater([0, 1, 2, 3, 4, 5]).step_by(2)

        print(eat.next())  # Some(0)
        print(eat.next())  # Some(2)
        print(eat.next())  # Some(4)
        print(eat.next())  # Option.none()
        ```
    """
    if step == 1:
        return self  # type: ignore
    return StepByEaterator(self, step)

take(n)

Creates an iterator that only yields the first n elements.

May be fewer than the requested amount.

Parameters:

Name Type Description Default
n int

Number of elements.

required
Source code in eaterate/core.py
def take(self, n: int, /) -> "TakeEaterator[T]":
    """Creates an iterator that only yields the first `n` elements.

    May be fewer than the requested amount.

    Args:
        n: Number of elements.
    """
    return TakeEaterator(self, n)

try_for_each(fn, _errhint=Exception)

Calls a falliable function on each element of this iterator.

Stops when one iteration has an error (exception) occurred.

Example

Let's assume you have a function defined for try_for_each that may fail, as well as an iterator. You'll notice that try_for_each gracefully catches the error, and returns it.

def nah(x: int):
    raise RuntimeError("hell nawh!")

# the iterator
eat = eater([1, 2, 3])

err = eat.try_for_each(nah)
if err is not None:
    print(err)  # hell nawh!
else:
    print('ok')

If needed, you can also provide the type checker with exception hints. If provided, only that exception will be caught.

eat.try_for_each(nah, RuntimeError)

Parameters:

Name Type Description Default
fn Callable

The function. Takes one parameter: an element.

required
_errhint Exception

Type hint that specifies what error may occur or be caught.

Exception
Source code in eaterate/core.py
def try_for_each(
    self, fn: Callable[[T], Any], _errhint: type[E] = Exception, /
) -> Union[E, None]:  # not to be confused with Option
    """Calls a falliable function on each element of this iterator.

    Stops when one iteration has an error (exception) occurred.

    Example:
        Let's assume you have a function defined for `try_for_each` that may fail, as well as
        an iterator. You'll notice that `try_for_each` gracefully catches the error, and returns it.
        ```python
        def nah(x: int):
            raise RuntimeError("hell nawh!")

        # the iterator
        eat = eater([1, 2, 3])

        err = eat.try_for_each(nah)
        if err is not None:
            print(err)  # hell nawh!
        else:
            print('ok')
        ```

        If needed, you can also provide the type checker with exception hints.
        If provided, only that exception will be caught.

        ```python
        eat.try_for_each(nah, RuntimeError)
        ```

    Args:
        fn (Callable): The function. Takes one parameter: an element.
        _errhint (Exception, optional): Type hint that specifies what error may occur or be caught.
    """
    while True:
        x = self.next()
        if x.is_none():
            break
        try:
            fn(x._unwrap())
        except _errhint as err:
            return err

windows(size)

Creates an iterator over overlapping subslices of length size.

Example
eat = eater([1, 2, 3, 4]).windows(2)

print(eat.next())  # Some([1, 2])
print(eat.next())  # Some([2, 3])
print(eat.next())  # Some([3, 4])
print(eat.next())  # Option.none()

When size is greater than the actual size of the original iterator, this immediately stops.

eat = eater([1, 2, 3]).windows(5)
print(eat.next())  # Option.none()
Source code in eaterate/core.py
def windows(self, size: int) -> "WindowsEaterator[T]":
    """Creates an iterator over overlapping subslices of length `size`.

    Example:
        ```python
        eat = eater([1, 2, 3, 4]).windows(2)

        print(eat.next())  # Some([1, 2])
        print(eat.next())  # Some([2, 3])
        print(eat.next())  # Some([3, 4])
        print(eat.next())  # Option.none()
        ```

        When `size` is greater than the *actual size* of the original iterator, this
        immediately stops.

        ```python
        eat = eater([1, 2, 3]).windows(5)
        print(eat.next())  # Option.none()
        ```
    """
    return WindowsEaterator(self, size)

zip(eat)

'Zips up' two iterators into a single iterator of pairs.

This returns a new iterator that will iterate over two other iterators, returning a tuple where the first element comes from the first iterator, and the second element comes from the second iterator.

Stops when either one of them has stopped.

This behaves like Python's built-in zip(), except only accepting one iterator only.

Examples:

(1) You can simply pass in two iterators.

eat = eater([0, 1, 2]).zip([1, 2, 3])

print(eat.next())  # Some((0, 1))
print(eat.next())  # Some((1, 2))
print(eat.next())  # Some((2, 3))
print(eat.next())  # Option.none()

(2) Sometimes their lengths don't match. It stops whenever one of the two iterators stops.

eat = eater([0, 1, 2]).zip([1, 2, 3, 4, 5])

print(eat.next())  # Some((0, 1))
print(eat.next())  # Some((1, 2))
print(eat.next())  # Some((2, 3))
print(eat.next())  # Option.none()

(3) When extracting more than two zipped iterators, beware of the (tuple) syntax.

eat = eater([0, 1, 2]).zip([2, 3, 4]).zip([4, 5, 6])

for (a, b), c in it:
    print(a, b, c)

Parameters:

Name Type Description Default
eat AutoIt[K]

The other iterator.

required
Source code in eaterate/core.py
def zip(self, eat: "AutoIt[K]", /) -> "ZipEaterator[T, K]":
    """'Zips up' two iterators into a single iterator of pairs.

    This returns a new iterator that will iterate over two other iterators, returning a tuple
    where the first element comes from the first iterator, and the second element comes from the second iterator.

    Stops when either one of them has stopped.

    This behaves like Python's built-in `zip()`, except only accepting one iterator only.

    Examples:

    (1) You can simply pass in two iterators.

    ```python
    eat = eater([0, 1, 2]).zip([1, 2, 3])

    print(eat.next())  # Some((0, 1))
    print(eat.next())  # Some((1, 2))
    print(eat.next())  # Some((2, 3))
    print(eat.next())  # Option.none()
    ```

    (2) Sometimes their lengths don't match. It stops whenever one of the two iterators stops.

    ```python
    eat = eater([0, 1, 2]).zip([1, 2, 3, 4, 5])

    print(eat.next())  # Some((0, 1))
    print(eat.next())  # Some((1, 2))
    print(eat.next())  # Some((2, 3))
    print(eat.next())  # Option.none()
    ```

    (3) When extracting more than two zipped iterators, beware of the `(tuple)` syntax.

    ```python
    eat = eater([0, 1, 2]).zip([2, 3, 4]).zip([4, 5, 6])

    for (a, b), c in it:
        print(a, b, c)
    ```

    Args:
        eat: The other iterator.
    """
    return ZipEaterator(self, eater(eat))