Generators em Python: O que É e Como Funciona | Python Brasil

# Função normal — carrega tudo na memória de uma vez
def quadrados_lista(n):
    resultado = []
    for i in range(n):
        resultado.append(i ** 2)
    return resultado

# Generator — gera um valor por vez, sob demanda
def quadrados_generator(n):
    for i in range(n):
        yield i ** 2

# A função retorna um objeto generator, sem executar nada ainda
gen = quadrados_generator(5)
print(type(gen))      # <class 'generator'>

print(next(gen))      # 0  — executa até o primeiro yield
print(next(gen))      # 1  — retoma do ponto de parada
print(next(gen))      # 4

# Iterando com for (chama next() automaticamente)
for quadrado in quadrados_generator(5):
    print(quadrado)   # 0, 1, 4, 9, 16

import sys

# Lista com 1 milhão de elementos
lista = [x for x in range(1_000_000)]
print(sys.getsizeof(lista))   # ~ 8.056.000 bytes (cerca de 8 MB)

# Generator equivalente
gen = (x for x in range(1_000_000))
print(sys.getsizeof(gen))     # 200 bytes (independente do tamanho!)

# Para arquivos grandes isso é ainda mais crítico
def ler_linhas(caminho):
    """Lê linhas de um arquivo gigante sem carregar tudo na memória."""
    with open(caminho, encoding='utf-8') as f:
        for linha in f:
            yield linha.strip()

# List comprehension — cria a lista completa imediatamente
lista = [x ** 2 for x in range(1000)]

# Generator expression — avaliação preguiçosa
gen = (x ** 2 for x in range(1000))

# Uso direto em funções que aceitam iteráveis
soma = sum(x ** 2 for x in range(1000))         # sem colchetes extras
maximo = max(len(s) for s in ["Python", "Go"])

def gen_a():
    yield 1
    yield 2

def gen_b():
    yield 3
    yield 4

# Sem yield from — verboso
def combinado_manual():
    for valor in gen_a():
        yield valor
    for valor in gen_b():
        yield valor

# Com yield from — limpo e idiomático
def combinado():
    yield from gen_a()
    yield from gen_b()
    yield from range(5, 8)

print(list(combinado()))  # [1, 2, 3, 4, 5, 6, 7]

# yield from com retorno — captura o valor de return do sub-generator
def sub():
    yield 10
    yield 20
    return "pronto"

def principal():
    resultado = yield from sub()
    print(f"Sub-generator retornou: {resultado}")
    yield 99

list(principal())  # Sub-generator retornou: pronto

def acumulador():
    total = 0
    while True:
        valor = yield total    # yield retorna total E recebe o próximo valor
        if valor is None:
            break
        total += valor

acc = acumulador()
next(acc)          # Inicializa o generator (avança até o primeiro yield)

print(acc.send(10))   # 10
print(acc.send(20))   # 30
print(acc.send(5))    # 35

# Exemplo prático: filtro configurável
def filtro_minimo():
    minimo = yield "Informe o mínimo:"
    while True:
        valor = yield
        if valor >= minimo:
            yield valor
        else:
            yield None

def ler_arquivo(caminho):
    """Etapa 1: lê linhas do arquivo."""
    with open(caminho, encoding='utf-8') as f:
        yield from f

def remover_cabecalho(linhas):
    """Etapa 2: pula a primeira linha (cabeçalho CSV)."""
    next(linhas)
    yield from linhas

def parsear_csv(linhas):
    """Etapa 3: converte cada linha em dicionário."""
    import csv
    leitor = csv.DictReader(linhas)  # csv.DictReader é iterável
    yield from leitor

def filtrar_ativos(registros):
    """Etapa 4: mantém apenas registros com status ativo."""
    for r in registros:
        if r.get('status') == 'ativo':
            yield r

def transformar(registros):
    """Etapa 5: normaliza campos."""
    for r in registros:
        r['nome'] = r['nome'].strip().title()
        r['valor'] = float(r.get('valor', 0))
        yield r

# Montando o pipeline — memória constante independente do tamanho do arquivo
def pipeline_etl(caminho):
    linhas = ler_arquivo(caminho)
    registros = parsear_csv(linhas)
    ativos = filtrar_ativos(registros)
    return transformar(ativos)

# Processa linha por linha, nunca carregando o arquivo completo
for registro in pipeline_etl('dados.csv'):
    salvar_no_banco(registro)

import itertools

# chain — encadeia múltiplos iteráveis
for item in itertools.chain([1, 2], [3, 4], [5]):
    print(item)   # 1, 2, 3, 4, 5

# islice — fatia um iterável (inclusive infinito)
def numeros_naturais():
    n = 0
    while True:
        yield n
        n += 1

primeiros_10 = list(itertools.islice(numeros_naturais(), 10))
# [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

# takewhile — pega enquanto a condição for verdadeira
menores_que_5 = list(itertools.takewhile(lambda x: x < 5, numeros_naturais()))
# [0, 1, 2, 3, 4]

# accumulate — acumula resultados (como reduce, mas lazy)
import operator
fatoriais = list(itertools.accumulate(range(1, 7), operator.mul))
# [1, 2, 6, 24, 120, 720]

# batched (Python 3.12+) — agrupa em lotes
lotes = list(itertools.batched(range(10), 3))
# [(0, 1, 2), (3, 4, 5), (6, 7, 8), (9,)]

import csv
import json
from datetime import datetime

def extrair_vendas(caminho_csv):
    """Extrai registros do arquivo CSV."""
    with open(caminho_csv, newline='', encoding='utf-8') as f:
        yield from csv.DictReader(f)

def transformar_vendas(registros):
    """Transforma e valida cada registro."""
    for r in registros:
        try:
            yield {
                'id': int(r['id']),
                'produto': r['produto'].strip(),
                'valor': float(r['valor']),
                'data': datetime.strptime(r['data'], '%Y-%m-%d').date(),
                'valido': float(r['valor']) > 0,
            }
        except (ValueError, KeyError):
            continue  # pula registros malformados

def filtrar_validos(registros):
    """Mantém apenas registros válidos."""
    return (r for r in registros if r['valido'])

def agrupar_por_produto(registros):
    """Acumula totais por produto."""
    totais = {}
    for r in registros:
        produto = r['produto']
        totais[produto] = totais.get(produto, 0) + r['valor']
    return totais

# Execução do pipeline
pipeline = filtrar_validos(
    transformar_vendas(
        extrair_vendas('vendas.csv')
    )
)

totais = agrupar_por_produto(pipeline)
print(json.dumps(totais, ensure_ascii=False, indent=2))

# Erro 1: tentar reutilizar um generator esgotado
gen = (x for x in range(3))
list(gen)   # [0, 1, 2]
list(gen)   # [] — generator já foi consumido!

# Solução: crie uma nova instância ou use uma função
def meu_gen():
    yield from range(3)

# Erro 2: esquecer de inicializar com next() antes de send()
def meu_coro():
    valor = yield

coro = meu_coro()
# coro.send(10)   # TypeError! Precisa chamar next() primeiro
next(coro)        # inicializa
coro.send(10)     # agora funciona

# Erro 3: usar return com valor em vez de yield (Python 2 thinking)
# Em Python 3, return em um generator encerra a iteração
# e o valor fica em StopIteration.value (acessado via yield from)