BehavioralPythonverifiedVerified
Iterator Pattern in Python
Provides a way to sequentially access elements of a collection without exposing its underlying representation.
How to Implement the Iterator Pattern in Python
1Step 1: Implement the collection as a Python iterable using __iter__
class NumberRange:
"""A range of numbers with a custom step, implementing the iterator protocol."""
def __init__(self, start: int, end: int, step: int = 1) -> None:
self._start = start
self._end = end
self._step = step
def __iter__(self) -> "RangeIterator":
return RangeIterator(self._start, self._end, self._step)2Step 2: Implement the iterator with __next__ and StopIteration
class RangeIterator:
def __init__(self, start: int, end: int, step: int) -> None:
self._current = start
self._end = end
self._step = step
def __iter__(self) -> "RangeIterator":
return self
def __next__(self) -> int:
if self._current > self._end:
raise StopIteration
value = self._current
self._current += self._step
return value3Step 3: Traverse the range using Python's for loop
number_range = NumberRange(1, 10, 2)
for num in number_range:
print(num) # 1, 3, 5, 7, 9
# Alternatively, use a generator for the same effect
def number_range_gen(start: int, end: int, step: int = 1):
current = start
while current <= end:
yield current
current += step
for num in number_range_gen(1, 10, 2):
print(num)"""Paginated API Iterator using async generators."""
import logging
from dataclasses import dataclass, field
from typing import Any, AsyncIterator, Callable, Awaitable, TypeVar
logger = logging.getLogger(__name__)
T = TypeVar("T")
# [step] Define page data structure and fetch function type
@dataclass(frozen=True)
class Page[T]:
items: list[T]
next_cursor: str | None
total_count: int
FetchPage = Callable[[str | None, int], Awaitable[Page[T]]]
# [step] Implement the async paginated iterator
class PaginatedIterator[T]:
def __init__(
self,
fetch_page: FetchPage[T],
page_size: int = 20,
max_items: int | None = None,
) -> None:
self._fetch_page = fetch_page
self._page_size = page_size
self._max_items = max_items
self._buffer: list[T] = []
self._cursor: str | None = None
self._done = False
self._fetched = 0
def __aiter__(self) -> "PaginatedIterator[T]":
return self
async def __anext__(self) -> T:
if self._max_items is not None and self._fetched >= self._max_items:
raise StopAsyncIteration
if not self._buffer:
if self._done:
raise StopAsyncIteration
page = await self._fetch_page(self._cursor, self._page_size)
self._buffer = list(page.items)
self._cursor = page.next_cursor
if page.next_cursor is None:
self._done = True
if not self._buffer:
raise StopAsyncIteration
self._fetched += 1
return self._buffer.pop(0)
# [step] Utility: collect all items from a paginated source
async def collect_all[T](
fetch_page: FetchPage[T],
*,
page_size: int = 20,
max_items: int | None = None,
) -> list[T]:
results: list[T] = []
async for item in PaginatedIterator(fetch_page, page_size, max_items):
results.append(item)
return results
# [step] Alternative: use an async generator for concise iteration
async def paginate[T](
fetch_page: FetchPage[T],
page_size: int = 20,
) -> AsyncIterator[T]:
cursor: str | None = None
while True:
page = await fetch_page(cursor, page_size)
for item in page.items:
yield item
if page.next_cursor is None:
break
cursor = page.next_cursorIterator Pattern Architecture
hourglass_empty
Rendering diagram...
lightbulb
Iterator Pattern in the Real World
“Think of a TV remote control. Whether your playlist is on a Blu-ray disc, a streaming service, or a USB drive, you use the same next and previous buttons to cycle through content. The remote (iterator interface) abstracts away the completely different internal mechanisms each media source uses to retrieve the next item.”