Introdução aos Generators com Comunicação Bidirecional
Generators em Python são funções que utilizam yield para produzir uma sequência de valores sob demanda. O que muitos desenvolvedores não sabem é que generators podem fazer muito mais: eles permitem comunicação bidirecional através dos métodos send(), throw() e close(). Essa capacidade transforma generators em máquinas de estado poderosas e elegantes, ideais para implementar corrotinas, processamento de fluxos de dados e controle fino de execução. Neste artigo, exploraremos como dominar essa funcionalidade avançada.
Comunicação Bidirecional: send(), throw() e close()
O método send()
O send() permite enviar valores para dentro de um generator enquanto ele está pausado em um yield. Quando um generator recebe um valor via send(), a expressão yield retorna esse valor, permitindo que a função processe e tome decisões baseadas nele.
def gerador_interativo():
print("Iniciando generator")
x = yield "Pronto para receber dados"
print(f"Recebi: {x}")
y = yield f"Valor processado: {x * 2}"
print(f"Novo valor: {y}")
yield "Finalizando"
# Uso
gen = gerador_interativo()
print(next(gen)) # Inicia o generator
print(gen.send(10)) # Envia 10, x recebe 10
print(gen.send(5)) # Envia 5, y recebe 5
Saída:
Iniciando generator
Pronto para receber dados
Recebi: 10
Valor processado: 20
Novo valor: 5
Finalizando
Os métodos throw() e close()
throw() injeta uma exceção dentro do generator, permitindo tratamento de erros sofisticado. close() encerra o generator injetando uma GeneratorExit.
def processador_resiliente():
try:
while True:
valor = yield
if valor is None:
print("Pulando valores nulos")
continue
print(f"Processando: {valor}")
except ValueError as e:
print(f"Erro capturado: {e}")
except GeneratorExit:
print("Generator encerrado corretamente")
gen = processador_resiliente()
next(gen)
gen.send(5)
gen.send(None)
gen.send(10)
try:
gen.throw(ValueError, ValueError("Dado inválido"))
except ValueError:
pass
gen.close()
Padrões Avançados: Corrotinas e Pipelines de Dados
Corrotinas para Processamento Assíncrono
Uma corrotina é um generator decorado com @coroutine (ou usando async/await moderno) que implementa um padrão produtor-consumidor. Elas são especialmente úteis para processar streams de dados com estado.
from functools import wraps
def coroutine(func):
"""Decorator que inicia o generator automaticamente"""
@wraps(func)
def wrapper(*args, **kwargs):
gen = func(*args, **kwargs)
next(gen) # Executa até o primeiro yield
return gen
return wrapper
@coroutine
def contador_acumulador():
total = 0
while True:
valor = yield total
if valor is not None:
total += valor
@coroutine
def filtro_pares():
while True:
valor = yield
if valor % 2 == 0:
print(f"Par encontrado: {valor}")
# Uso em pipeline
acc = contador_acumulador()
print(acc.send(5)) # 5
print(acc.send(3)) # 8
print(acc.send(2)) # 10
Pipeline de Transformação
Pipelines combinam múltiplos generators para processar dados em etapas. Cada estágio recebe dados, processa e passa adiante.
@coroutine
def produtor(dados):
for item in dados:
print(f"[Produtor] Enviando: {item}")
yield item
@coroutine
def transformador(proximo):
while True:
valor = yield
resultado = valor * 2
print(f"[Transformador] {valor} → {resultado}")
proximo.send(resultado)
@coroutine
def consumidor():
while True:
valor = yield
print(f"[Consumidor] Recebeu: {valor}\n")
# Montando pipeline
dados = [1, 2, 3, 4, 5]
cons = consumidor()
trans = transformador(cons)
for item in dados:
trans.send(item)
trans.close()
Controle de Fluxo e Máquinas de Estado
Implementando Máquinas de Estado
Generators permitem implementar máquinas de estado de forma elegante, mantendo o estado interno sem necessidade de classes complexas.
def maquina_estados():
estado = "inicio"
while True:
evento = yield estado
if estado == "inicio":
if evento == "conectar":
estado = "conectado"
elif evento == "erro":
estado = "erro"
elif estado == "conectado":
if evento == "enviar":
estado = "enviando"
elif evento == "desconectar":
estado = "desconectado"
elif estado == "enviando":
if evento == "sucesso":
estado = "conectado"
elif evento == "falha":
estado = "erro"
elif estado == "erro":
if evento == "reconectar":
estado = "inicio"
# Simulação
maq = maquina_estados()
print(next(maq)) # inicio
print(maq.send("conectar")) # conectado
print(maq.send("enviar")) # enviando
print(maq.send("sucesso")) # conectado
print(maq.send("desconectar")) # desconectado
Delegação com yield from
yield from simplifica a delegação para sub-generators, permitindo composição elegante:
def gerador_a():
yield 1
yield 2
def gerador_b():
yield 3
yield 4
def gerador_delegado():
yield from gerador_a()
yield from gerador_b()
yield 5
for valor in gerador_delegado():
print(valor) # 1, 2, 3, 4, 5
Conclusão
Os três aprendizados principais que consolidam seu domínio sobre generators avançados são: (1) Comunicação bidirecional via send(), throw() e close() transforma generators em estruturas ativas capazes de processar dados contextualizados; (2) Padrões como corrotinas e pipelines habilitam processamento de streams elegante e modular, reduzindo complexidade em código de produção; (3) Máquinas de estado implementadas com generators oferecem alternativa limpa e performática à orientação a objetos pesada, ideal para sistemas com múltiplos estados e transições.
Pratique construindo pequenos projetos: um parser de eventos, um simulador de fila de processamento ou um sistema de logging com backpressure. A prática solidifica a intuição sobre quando e como usar cada recurso.