Итераторы через __iter__ и __next__. Генераторы и yield from. Lazy evaluation и бесконечные последовательности.
Итераторы превращают коллекции в потоки. Генераторы превращают функции в итераторы.
Iterator — поведенческий паттерн, предоставляющий способ последовательного доступа к элементам составного объекта без раскрытия его внутреннего представления.
# Протокол итератора
class Iterator:
def __iter__(self) -> "Iterator":
"""Возвращает сам итератор"""
return self
def __next__(self) -> Any:
"""Возвращает следующий элемент или StopIteration"""
...
# Протокол iterable
class Iterable:
def __iter__(self) -> Iterator:
"""Возвращает итератор"""
...# ✅ Итератор для обхода вложенных списков
from typing import List, Any, Optional
class FlatIterator:
"""Итератор для плоского обхода вложенных списков"""
def __init__(self, nested_list: List):
self._nested = nested_list
self._index = 0
self._sub_iter: Optional[Iterator] = None
def __iter__(self) -> "FlatIterator":
return self
def __next__(self) -> Any:
while True:
# Если есть активный под-итератор
if self._sub_iter:
try:
return next(self._sub_iter)
except StopIteration:
self._sub_iter = None
continue
# Если индекс за границами
if self._index >= len(self._nested):
raise StopIteration
item = self._nested[self._index]
self._index += 1
# Если элемент список — создаём под-итератор
if isinstance(item, list):
self._sub_iter = iter(item)
else:
return item
# Использование
nested = [1, [2, 3], 4, [5, [6, 7]]]
for item in FlatIterator(nested):
print(item) # 1, 2, 3, 4, 5, [6, 7]# ✅ Итератор Фибоначчи
class FibonacciIterator:
def __init__(self, max_value: int = None):
self._max = max_value
self._a = 0
self._b = 1
def __iter__(self) -> "FibonacciIterator":
self._a = 0
self._b = 1
return self
def __next__(self) -> int:
result = self._a
if self._max is not None and result > self._max:
raise StopIteration
self._a, self._b = self._b, self._a + self._b
return result
# Использование
for fib in FibonacciIterator(100):
print(fib) # 0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89# ✅ Генератор вместо класса-итератора
def fibonacci(max_value: int = None):
"""Генератор Фибоначчи"""
a, b = 0, 1
while True:
if max_value is not None and a > max_value:
return
yield a
a, b = b, a + b
# Использование
for fib in fibonacci(100):
print(fib)
# Или вручную
gen = fibonacci(10)
print(next(gen)) # 0
print(next(gen)) # 1
print(next(gen)) # 1Преимущество: Генератор автоматически реализует протокол итератора. yield сохраняет состояние между вызовами.
# ✅ Двусторонняя связь через send
def accumulator():
"""Генератор, принимающий значения"""
total = 0
while True:
value = yield total
if value is None:
break
total += value
# Использование
acc = accumulator()
next(acc) # 0 (запуск до первого yield)
print(acc.send(5)) # 5 (total = 5)
print(acc.send(3)) # 8 (total = 8)
print(acc.send(10)) # 18 (total = 18)
acc.send(None) # Завершение# ✅ Исключения в генераторе
def safe_divide():
"""Генератор с обработкой исключений"""
while True:
try:
a = yield
b = yield
result = a / b
print(f"Result: {result}")
except ZeroDivisionError:
print("Division by zero!")
# Использование
gen = safe_divide()
next(gen) # Запуск
gen.send(10) # a = 10
gen.send(2) # Result: 5.0
gen.send(10) # a = 10
gen.send(0) # Division by zero!# ✅ yield from упрощает вложенные генераторы
def chain(*iterables):
"""Цепочка итераторов"""
for iterable in iterables:
yield from iterable
# Использование
result = list(chain([1, 2], [3, 4], [5, 6]))
# [1, 2, 3, 4, 5, 6]
# Без yield from пришлось бы:
def chain_verbose(*iterables):
for iterable in iterables:
for item in iterable:
yield item# ✅ Рекурсивный обход дерева
def flatten(nested):
"""Рекурсивное выравнивание списка"""
for item in nested:
if isinstance(item, list):
yield from flatten(item)
else:
yield item
# Использование
nested = [1, [2, 3], [4, [5, 6]], 7]
print(list(flatten(nested))) # [1, 2, 3, 4, 5, 6, 7]# ✅ Ленивое чтение файла
def read_lines(filename: str):
"""Генератор для чтения файла по строкам"""
with open(filename) as f:
for line in f:
yield line.rstrip()
# Использование — файл не загружается целиком
for line in read_lines("huge_file.txt"):
process(line)# ✅ Конвейер через генераторы
def filter_positive(numbers):
for n in numbers:
if n > 0:
yield n
def square(numbers):
for n in numbers:
yield n ** 2
def take(n, numbers):
"""Берёт первые n элементов"""
for i, x in enumerate(numbers):
if i >= n:
break
yield x
# Конвейер
numbers = range(-10, 10)
pipeline = take(5, square(filter_positive(numbers)))
print(list(pipeline)) # [1, 4, 9, 16, 25]# ✅ Пагинация через генератор
from typing import List, Any, Callable
def paginate(fetch_func: Callable[[int, int], List[Any]],
page_size: int = 100):
"""Генератор для постраничной загрузки"""
page = 0
while True:
items = fetch_func(page, page_size)
if not items:
break
yield from items
page += 1
# Использование
def fetch_users(page: int, size: int) -> List[dict]:
# Имитация API
if page >= 3:
return []
return [{"id": page * size + i} for i in range(size)]
for user in paginate(fetch_users, page_size=100):
print(user["id"]) # 0-299# ✅ Бесконечные генераторы
def count(start: int = 0, step: int = 1):
"""Бесконечный счётчик"""
n = start
while True:
yield n
n += step
def repeat(value):
"""Бесконечный повтор"""
while True:
yield value
def cycle(iterable):
"""Бесконечный цикл по iterable"""
saved = []
for item in iterable:
yield item
saved.append(item)
while saved:
yield from saved
# Использование
from itertools import islice
print(list(islice(count(10), 5))) # [10, 11, 12, 13, 14]
print(list(islice(repeat("X"), 3))) # ['X', 'X', 'X']
print(list(islice(cycle([1, 2]), 5))) # [1, 2, 1, 2, 1]# ✅ Разбиение на пакеты
def batch(iterable, size: int):
"""Разбивает итератор на пакеты"""
batch = []
for item in iterable:
batch.append(item)
if len(batch) >= size:
yield batch
batch = []
if batch:
yield batch
# Использование
data = range(10)
for batch_items in batch(data, 3):
print(batch_items) # [0,1,2], [3,4,5], [6,7,8], [9]from contextlib import contextmanager
@contextmanager
def transaction(db):
"""Транзакция через генератор"""
try:
yield db # Код в with выполняется здесь
db.commit()
except Exception:
db.rollback()
raise
# Использование
with transaction(db) as db:
db.execute("INSERT ...")# ✅ Генератор как корутина
def coroutine():
"""Генератор-корутина"""
while True:
value = yield
print(f"Received: {value}")
# Использование
coro = coroutine()
next(coro) # Запуск
coro.send("Hello") # Received: HelloИзучите тему Mediator и Event Bus для централизованного взаимодействия объектов.
Далее: Mediator и Event Bus