Fundamentos do Padrão Iterator

O padrão Iterator é um padrão de projeto comportamental essencial que permite a travessia sequencial de elementos de um objeto agregado sem expor sua representação interna. No ecossistema Python, esse conceito é nativo e onipresente, servindo como base para construções como loops for, expressões geradoras e compreensões de lista.

A adoção deste padrão traz benefícios arquitetônicos significativos:

• Desacoplamento de responsabilidades: Separa a lógica de armazenamento de dados da lógica de navegação.
• Polimorfismo de travessia: Permite que uma mesma coleção seja percorrida de diferentes maneiras utilizando iteradores distintos.
• Avaliação preguiçosa (Lazy Evaluation): Computa e retorna elementos sob demanda, otimizando o consumo de memória em fluxos de dados massivos.

A biblioteca PyPattyrn fornece uma base sólida para este padrão, definindo as abstrações necessárias para que desenvolvedores construam seus próprios iteradores personalizados de forma padronizada.

Arquitetura Base do Iterador

A implementação do padrão no PyPattyrn baseia-se em duas abstrações principais. Para fins de clareza e adaptação, utilizaremos nomenclaturas que refletem diretamente o protocolo de iteração do Python.

Abstração da Coleção

A classe base define o contrato para qualquer objeto que possa ser iterado. Utilizando o módulo abc do Python, garantimos que as subclasses implementem o método de extração do próximo elemento.

from abc import ABC, abstractmethod

class DataCollection(ABC):
    """
    Define a interface para objetos que podem ser percorridos.
    """
    @abstractmethod
    def __next__(self):
        """
        Deve retornar o próximo item ou lançar StopIteration.
        """
        pass

Classe do Iterador

O iterador concreto encapsula a lógica de controle do estado da travessia, delegando a obtenção dos dados para a coleção alvo.

class DataIterator:
    """
    Implementa o protocolo de iteração delegando a busca à coleção.
    """
    def __init__(self, source: DataCollection):
        self.source = source

    def __iter__(self):
        return self

    def __next__(self):
        return self.source.__next__()

Construindo um Iterador Personalizado

Para demonstrar a aplicação prática, vamos criar uma sequência numérica que avança com um passo (step) definido, em vez de um simples contador unitário.

Passo 1: Definir a Coleção de Dados

class SteppedSequence(DataCollection):
    def __init__(self, start_val: int, limit_val: int, step_val: int):
        self.cursor = start_val
        self.limit = limit_val
        self.step = step_val

    def __next__(self):
        if self.cursor >= self.limit:
            raise StopIteration
        
        current_item = self.cursor
        self.cursor += self.step
        return current_item

Passo 2: Encapsular no Iterador

class SteppedIterator(DataIterator):
    def __init__(self):
        # Inicializa com uma sequência de 0 a 50, pulando de 10 em 10
        super().__init__(SteppedSequence(0, 50, 10))

Passo 3: Executar a Travessia

sequence_runner = SteppedIterator()
for value in sequence_runner:
    print(f"Processando item: {value}")
    # Saída: 0, 10, 20, 30, 40

Cenários de Aplicação Avançada

1. Processamento de Fluxos de Logs

Iteradores são ideais para ler arquivos de log gigantes linha por linha, evitendo o carregamento do arquivo inteiro na RAM.

class LogStream(DataCollection):
    def __init__(self, file_path: str):
        self.file_handler = open(file_path, 'r', encoding='utf-8')
    
    def __next__(self):
        raw_line = self.file_handler.readline()
        if not raw_line:
            self.file_handler.close()
            raise StopIteration
        return raw_line.strip()

# Consumo eficiente de memória
# for log_entry in DataIterator(LogStream("server_access.log")):
#     parse_and_store(log_entry)

2. Geração de Sequências Infinitas

Como a avaliação é preguiçosa, podemos modelar séries matemáticas infinitas, como potências de base 2.

class PowerOfTwoStream(DataCollection):
    def __init__(self):
        self.current_power = 1
    
    def __next__(self):
        result = self.current_power
        self.current_power *= 2
        return result

# Extraindo apenas os primeiros 5 elementos
# pow_iterator = DataIterator(PowerOfTwoStream())
# for _ in range(5):
#     print(next(pow_iterator))

3. Encadeamento de Múltiplas Fontes

É possível criar um iterador que consome múltiplas coleções sequencialmente, transparentemente para o cliente.

class ChainedDataIterator(DataIterator):
    def __init__(self, collections_list: list):
        self.active_streams = [DataIterator(c) for c in collections_list]
        self.current_stream_idx = 0
    
    def __next__(self):
        while self.current_stream_idx < len(self.active_streams):
            try:
                return next(self.active_streams[self.current_stream_idx])
            except StopIteration:
                self.current_stream_idx += 1
        raise StopIteration

# Uso:
# combined = ChainedDataIterator([SteppedSequence(0, 5, 1), SteppedSequence(10, 15, 1)])
# for item in combined: print(item)

Validação e Testes Unitários

Garantir que o protocolo de iteração esteja correto exige testes que validem tanto a extração de valores quanto o comportamento de exaustão (StopIteration).

import unittest

class TestSteppedIteratorBehavior(unittest.TestCase):
    
    def setUp(self):
        self.target_iterator = SteppedIterator()

    def test_sequence_extraction(self):
        expected_values = [0, 10, 20, 30, 40]
        extracted_values = [next(self.target_iterator) for _ in range(5)]
        self.assertListEqual(expected_values, extracted_values)

    def test_iteration_exhaustion(self):
        # Consome todos os elementos disponíveis
        list(self.target_iterator) 
        
        # A próxima chamada deve falhar
        with self.assertRaises(StopIteration):
            next(self.target_iterator)

    def test_native_loop_compatibility(self):
        self.target_iterator = SteppedIterator()
        count = 0
        for _ in self.target_iterator:
            count += 1
        self.assertEqual(count, 5)