The 100 problems, solved / Iterators, streams and counters / 013 · Design a peeking iterator

Design a peeking iterator

Wrap any iterator so peek() shows the next item without consuming it.

Concepts this design leans on

01The design, in the six-step framework

1. Requirements and use cases

An easy-looking wrapper that Google uses to see whether you understand iterator contracts, laziness and the awkward cases: peek before next, next before peek, and an underlying source that yields null-like values.

  • A caller wraps any iterator and gets has_next, next and a new peek.
  • peek returns the next element without consuming it; repeated peeks return the same element.
  • next after a peek returns the peeked element and advances.
  • Wrapping is lazy: the underlying iterator is advanced only when needed.
  • The wrapper is generic over the element type, including elements that are themselves None.

Out of scope: thread safety (iterators are single-consumer by contract), multi-element look-ahead beyond one, and rewinding. Non-functional constraints: O(1) per call with one element of buffer; the wrapped iterator is pulled at most once per element; and exhausting behaviour matches the language's iterator protocol so the wrapper can be used in a plain loop.

2. Core entities and relationships

  • PeekingIterator: the wrapper; holds the source and a one-slot buffer with an explicit "buffer is filled" flag.
  • Source: any object with __next__; the wrapper never inspects what it is.
  • Buffer: the single look-ahead slot, distinguished from "empty" by a flag rather than by a sentinel value so None elements work.

The wrapper aggregates its source: it does not own or copy it, and it is the only reader once wrapping starts. The buffer is two fields, not a nullable value, because a nullable value cannot represent an element that is itself None; this small decision is the entire difference between a correct and a subtly broken wrapper. The shape is the Decorator pattern: same interface in, same interface out, one capability added.

WrapperSourcewrapsPeekingIterator[T]source: Iterator[T]buffered: boolbuffer: T | Nonepeek() -> T__next__() -> Thas_next() -> bool__iter__()Iterator[T]__next__() -> T

3. Interfaces and key methods

from typing import Generic, Iterator, TypeVar

T = TypeVar("T")


class PeekingIterator(Generic[T]):
    def __init__(self, source: Iterator[T]) -> None:
        """Takes ownership of reading from source; nothing is pulled yet."""

    def peek(self) -> T:
        """Next element without consuming it; raises StopIteration if exhausted."""

    def __next__(self) -> T:
        """The buffered element if any, else pulls from source; raises StopIteration when done."""

    def has_next(self) -> bool:
        """True if another element exists; may pull one element into the buffer."""

    def __iter__(self) -> "PeekingIterator[T]": ...

Invariants: when buffered is true, buffer holds the next element and the source has already been advanced past it; when false, buffer is meaningless; and the source is never advanced while the buffer is full. Together these guarantee each element is pulled exactly once.

4. Design patterns and the choices behind them

Decorator over the iterator protocol

The force is adding one method to an existing interface without changing the callers. A Decorator wraps and re-exposes __next__, so code that expects a plain iterator keeps working, and code that knows about the wrapper can peek. The rejected alternative is reading the whole source into a list, which is O(n) memory and breaks for infinite sources, and it is not lazy, which the requirement names explicitly. This is the standard iterator design move.

A flag, not a sentinel

The force is None as a legitimate element. Using buffer is None to mean "empty" makes the wrapper skip or double-yield a None in the source. The separate boolean costs one byte and removes the bug class. In statically typed languages the same issue appears as "the generic T might be null"; the follow-up about generics in the reported interviews is this point.

Exhaustion is a protocol, not a flag

The force is interoperability. Python signals the end of an iterator by raising StopIteration from __next__; Java and C# use a boolean query. The wrapper follows the host language exactly so a plain for loop over it just works, and it adds has_next for callers who prefer the query style. Mixing the two, for example returning None at the end, would collide with the None-element case above and would silently end loops early. Catching StopIteration only inside _fill, and nowhere else, keeps the one place where the protocol is translated into a boolean.

has_next may pull

The force is that most iterator protocols have no cheap "is there more" without reading. The honest implementation of has_next fills the buffer and returns whether it succeeded. Document that it is not a pure query; method design is about telling the caller which calls do work.

5. Core flow and code

Flow for peek(); peek(); next(); next() over source [1, 2]:

  1. First peek: buffer empty, pull 1 from source, store it, set the flag, return 1.
  2. Second peek: flag set, return the buffered 1 without touching the source.
  3. First next: flag set, clear it and return 1.
  4. Second next: flag clear, pull 2 directly from the source and return it. A third next raises StopIteration from the source.
from typing import Generic, Iterator, List, Optional, TypeVar

T = TypeVar("T")


class PeekingIterator(Generic[T]):
    def __init__(self, source: Iterator[T]) -> None:
        self._source = source
        self._buffered = False
        self._buffer: Optional[T] = None

    def _fill(self) -> bool:
        """Ensure the buffer holds the next element; False if the source is exhausted."""
        if self._buffered:
            return True
        try:
            self._buffer = next(self._source)
        except StopIteration:
            return False
        self._buffered = True
        return True

    def peek(self) -> T:
        if not self._fill():
            raise StopIteration
        return self._buffer  # type: ignore[return-value]

    def __next__(self) -> T:
        if self._buffered:
            self._buffered = False
            value, self._buffer = self._buffer, None
            return value  # type: ignore[return-value]
        return next(self._source)

    def has_next(self) -> bool:
        return self._fill()

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


class LookaheadIterator(Generic[T]):
    """Extension: peek k ahead with a bounded deque; k = 1 degenerates to PeekingIterator."""

    def __init__(self, source: Iterator[T], k: int) -> None:
        from collections import deque
        self._source, self._k = source, k
        self._buf: "deque[T]" = deque()

    def _fill(self, n: int) -> None:
        while len(self._buf) < n:
            try:
                self._buf.append(next(self._source))
            except StopIteration:
                break

    def peek(self, offset: int = 0) -> T:
        if offset >= self._k:
            raise ValueError("offset beyond lookahead window")
        self._fill(offset + 1)
        if offset >= len(self._buf):
            raise StopIteration
        return self._buf[offset]

    def __next__(self) -> T:
        self._fill(1)
        if not self._buf:
            raise StopIteration
        return self._buf.popleft()

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


if __name__ == "__main__":
    p = PeekingIterator(iter([1, 2]))
    assert p.peek() == 1 and p.peek() == 1 and next(p) == 1 and next(p) == 2
    assert not p.has_next()
    q = PeekingIterator(iter([None, 3]))          # None is a real element
    assert q.has_next() and q.peek() is None and next(q) is None and next(q) == 3
    r = PeekingIterator(iter([7, 8, 9]))
    assert next(r) == 7 and r.peek() == 8 and list(r) == [8, 9]   # next before peek, then plain loop
    la = LookaheadIterator(iter("abcd"), 2)
    assert la.peek(1) == "b" and next(la) == "a" and la.peek(1) == "c"

6. Extensions, concurrency and what interviewers probe

  • Peek before next, next before peek. Both orders are tested in the code; the second is the one candidates forget, because next must bypass an empty buffer and pull directly.
  • None elements. The flag makes them work; a sentinel object (_EMPTY = object()) is an equally good alternative in Python and worth mentioning.
  • Generics follow-up. The wrapper is parameterised over T and never compares elements, so any type works, including unhashable ones.
  • k-element look-ahead. The deque version above; memory O(k), still lazy.
  • Concurrency. Iterators are single-consumer; if two threads share one, wrap the whole object in a lock or give each thread its own via a tee.
  • What to test first. The four-call sequence in section 5, then a source of exactly one element where peek, has_next and next are called in every order; small inputs expose every branch of the buffer flag.
  • Common mistake. Calling has_next on the source in the constructor or caching elements eagerly, which breaks laziness and infinite sources.

02Where this question was reported

  • LeetCode Discuss · LeetCode 284 (Medium), Google-tagged, freq 18.2
  • evelynn.gitbooks.io · Google-tagged study collection; test peek-before-next and next-before-peek; generics follow-up
  • evelynn.gitbooks.io · Zigzag iterator; 'cyclic' once >2 lists; queue of sub-iterators
  • Taro · Taro claims Google/Meta
  • Medium · 'Real Google' peeking iterator write-up