Generadores avanzados en Python

Los generadores permiten crear secuencias de datos de forma eficiente, sin cargar todo en memoria. En esta guía aprenderás técnicas avanzadas como generadores infinitos, pipelines, delegación con yield from, comunicación bidireccional con send() y patrones profesionales usados en sistemas de streaming.

1. Generadores infinitos

Un generador puede producir valores infinitamente sin consumir memoria.

def contador():
    n = 0
    while True:
        yield n
        n += 1

gen = contador()
print(next(gen))
print(next(gen))
print(next(gen))

2. Generadores como pipelines

Puedes encadenar generadores para crear flujos de datos eficientes.

def numeros():
    for n in range(1, 10):
        yield n

def cuadrados(gen):
    for n in gen:
        yield n * n

def pares(gen):
    for n in gen:
        if n % 2 == 0:
            yield n

pipeline = pares(cuadrados(numeros()))
print(list(pipeline))

3. yield from

yield from delega la generación de valores a otro generador.

def subgen():
    yield 1
    yield 2
    yield 3

def gen():
    yield from subgen()
    yield 4

print(list(gen()))

4. Comunicación bidireccional con send()

Los generadores pueden recibir datos desde el exterior usando send().

def acumulador():
    total = 0
    while True:
        x = yield total
        total += x

gen = acumulador()
print(next(gen))      # inicia
print(gen.send(5))    # total = 5
print(gen.send(10))   # total = 15

5. Generadores como corutinas ligeras

Antes de async/await, los generadores se usaban como corutinas.

def procesar():
    while True:
        dato = yield
        print("Procesando:", dato)

p = procesar()
next(p)
p.send("mensaje 1")
p.send("mensaje 2")

6. Ejercicios

def pares():
    n = 0
    while True:
        yield n
        n += 2

def unir(a, b):
    yield from a
    yield from b

def comandos():
    while True:
        cmd = yield
        print("CMD:", cmd)

7. Mini‑proyecto: Stream de datos con generadores

def leer_lineas(ruta):
    with open(ruta) as f:
        for linea in f:
            yield linea.strip()

def filtrar_error(gen):
    for linea in gen:
        if "ERROR" in linea:
            yield linea

def mostrar(gen):
    for linea in gen:
        print(">>", linea)

pipeline = mostrar(filtrar_error(leer_lineas("logs.txt")))

Siguiente guía