Generadores e iteradores en Python: yield, lazy evaluation y memoria eficiente
📋 Contenido de esta lección
- El problema que resuelven los generadores
- Iteradores: iter(), next() y StopIteration
- Protocolo de iteración: __iter__ y __next__
- yield: pausar y reanudar una función
- Expresiones generadoras: comprensiones lazy
- yield from: delegar en otro generador
- send() y throw(): comunicación bidireccional
- itertools: la caja de herramientas de iteración
- Memoria y rendimiento: cuándo usar generadores
- Programa completo: pipeline de procesamiento de logs
- Errores clásicos
- Preguntas frecuentes
The key question is not whether to use an object, but what that object's interface should be
«La clave no es si usar un objeto, sino cuál debe ser su interfaz»
Barbara Liskov · Premio Turing 2008 · 1939 –
Barbara Liskov definió el principio de sustitución que lleva su nombre: si una interfaz promete cierto comportamiento, cualquier implementación de esa interfaz debe respetarlo. El protocolo de iteración de Python es precisamente eso: cualquier objeto que implemente
__iter__() y __next__() puede usarse en un for, en list(), en sum(). No importa si es una lista, un fichero, un generador o una clase propia. Liskov habría reconocido el diseño: una interfaz mínima, universalmente respetada, que hace intercambiables millones de objetos distintos.
Un generador produce valores uno a uno, bajo demanda, en lugar de calcularlos todos de golpe y guardarlos en memoria. Esta evaluación perezosa (lazy evaluation) permite trabajar con secuencias enormes —o incluso infinitas— con memoria constante, y escribir pipelines de transformación de datos que se componen como piezas de un puzzle. yield es la palabra clave que lo hace posible.
🔄 Iteradores: iter(), next() y StopIteration
# Python usa iteradores en todos lados: for, list(), sum(), zip(), map()…
# Entender el protocolo permite crear tus propios objetos iterables.
# iter() → obtiene el iterador de un iterable
# next() → obtiene el siguiente elemento del iterador
lista = [10, 20, 30]
it = iter(lista) # crea el iterador
next(it) # 10
next(it) # 20
next(it) # 30
next(it) # StopIteration ← señal de que se agotó
# Valor por defecto en next() — evita la excepción
next(it, "agotado") # 'agotado' ← no lanza StopIteration
# Un bucle for es exactamente esto, pero automático:
for x in [10, 20, 30]:
print(x)
# Python hace internamente:
# _it = iter([10, 20, 30])
# while True:
# try: x = next(_it)
# except StopIteration: break
# print(x)
# Los iteradores son ellos mismos iterables (iter(it) devuelve it)
it2 = iter(lista)
iter(it2) is it2 # True ← los iteradores implementan __iter__ devolviendo self
# Tipos que son iterables (tienen __iter__) pero NO son iteradores por sí mismos:
# list, tuple, str, dict, set, range → iter() crea un iterador nuevo cada vez
# Tipos que son iteradores: map(), filter(), zip(), enumerate(), generadores
# Diferencia clave: los iteradores son de un solo uso
it_a = iter([1, 2, 3])
list(it_a) # [1, 2, 3]
list(it_a) # [] ← ya se agotó
lista = [1, 2, 3]
list(lista) # [1, 2, 3]
list(lista) # [1, 2, 3] ← la lista se puede recorrer de nuevo
🔧 Protocolo de iteración: __iter__ y __next__
# Puedes hacer iterable cualquier clase implementando el protocolo
class ContadorRegresivo:
"""Itera desde n hasta 1 (inclusive)."""
def __init__(self, inicio):
self.inicio = inicio
self.actual = inicio
def __iter__(self):
"""Devuelve el iterador — en este caso, el propio objeto."""
self.actual = self.inicio # reinicia para poder reusar
return self
def __next__(self):
"""Devuelve el siguiente valor o lanza StopIteration."""
if self.actual <= 0:
raise StopIteration
valor = self.actual
self.actual -= 1
return valor
# Uso exactamente igual que cualquier iterable
cuenta = ContadorRegresivo(5)
for n in cuenta:
print(n, end=" ") # 5 4 3 2 1
list(cuenta) # [5, 4, 3, 2, 1] ← funciona de nuevo (reinicia en __iter__)
sum(cuenta) # 15
max(cuenta) # 5
# Rango personalizado con paso flotante (range() solo acepta enteros)
class RangoFlotante:
def __init__(self, inicio, fin, paso=1.0):
self.inicio = inicio
self.fin = fin
self.paso = paso
def __iter__(self):
actual = self.inicio
while actual < self.fin:
yield round(actual, 10)
actual += self.paso
# usar yield dentro de __iter__ convierte el método en un generador
for x in RangoFlotante(0, 1, 0.25):
print(x) # 0, 0.25, 0.5, 0.75
# Clase con __iter__ usando yield — la forma más concisa
class Fibonacci:
def __init__(self, limite):
self.limite = limite
def __iter__(self):
a, b = 0, 1
while a <= self.limite:
yield a
a, b = b, a + b
list(Fibonacci(100)) # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89]
⚡ yield: pausar y reanudar una función
# Una función con yield es una función generadora.
# Al llamarla NO ejecuta el cuerpo: devuelve un objeto generador.
# Cada next() reanuda desde donde se quedó hasta el siguiente yield.
def contar_hasta(n):
print("Inicio del generador")
for i in range(1, n + 1):
print(f" → antes de yield {i}")
yield i # pausa aquí, devuelve i
print(f" ← después de yield {i}")
print("Generador agotado")
gen = contar_hasta(3) # no ejecuta nada todavía
print(type(gen)) #
next(gen) # "Inicio del generador" + "→ antes de yield 1" → 1
next(gen) # "← después de yield 1" + "→ antes de yield 2" → 2
next(gen) # "← después de yield 2" + "→ antes de yield 3" → 3
next(gen) # "← después de yield 3" + "Generador agotado" → StopIteration
# Generador de números de Fibonacci (infinito)
def fibonacci():
a, b = 0, 1
while True: # ← bucle infinito — el generador nunca se agota
yield a
a, b = b, a + b
gen_fib = fibonacci()
[next(gen_fib) for _ in range(10)] # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]
# Generador de líneas de un fichero grande (sin cargar todo en memoria)
def leer_lineas(ruta):
with open(ruta, encoding="utf-8") as f:
for linea in f:
yield linea.rstrip("\n")
# Solo una línea en memoria en cada momento, aunque el fichero tenga 10 GB
for linea in leer_lineas("/var/log/syslog"):
if "ERROR" in linea:
print(linea)
# yield también puede devolver None implícitamente
def generador_con_return(n):
for i in range(n):
yield i
return "terminado" # el valor de return se pasa en StopIteration.value
gen = generador_con_return(3)
try:
while True:
print(next(gen))
except StopIteration as e:
print(f"Valor de retorno: {e.value}") # 'terminado'
💡 Lista vs generador:
[x**2 for x in range(10_000_000)] crea 10 millones de enteros en memoria (~80 MB). (x**2 for x in range(10_000_000)) crea un generador que ocupa ~120 bytes independientemente del tamaño. Si solo necesitas recorrerlo una vez, el generador es casi siempre la mejor opción.
🔵 Expresiones generadoras: comprensiones lazy
# Sintaxis: (expresión for x in iterable if condición)
# Igual que una comprensión de lista, pero con () en vez de []
# No calcula nada hasta que alguien consume el generador
# Lista: construye todo en memoria
cuadrados_lista = [x**2 for x in range(10)] # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
# Generador: lazy, un elemento a la vez
cuadrados_gen = (x**2 for x in range(10)) #
# Tamaño en memoria
import sys
sys.getsizeof([x**2 for x in range(1000)]) # ~8856 bytes
sys.getsizeof((x**2 for x in range(1000))) # ~208 bytes ← constante
# Pasar a funciones que consumen iterables — sin paréntesis extra si es el único arg
sum(x**2 for x in range(100)) # 328350
max(len(p) for p in ["hola", "mundo", "python"]) # 6
any(x > 5 for x in [1, 3, 7, 2]) # True
all(x > 0 for x in [1, 3, 7, 2]) # True
# Con condición
pares = (x for x in range(20) if x % 2 == 0)
grandes = (x**2 for x in range(100) if x**2 > 50)
# Anidar expresiones generadoras — pipeline lazy
numeros = range(1_000_000)
pipeline = (
x * 2
for x in numeros
if x % 3 == 0 # solo múltiplos de 3
)
# Todavía no calculó nada — evalúa bajo demanda:
next(pipeline) # 0
next(pipeline) # 6
next(pipeline) # 12
# Comparativa de memoria para procesar un CSV grande
import csv
# ❌ Todo en memoria a la vez
with open("ventas.csv") as f:
datos = [row for row in csv.DictReader(f)] # 100.000 filas en RAM
total = sum(float(row["importe"]) for row in datos)
# ✅ Streaming — solo una fila en memoria
with open("ventas.csv") as f:
total = sum(
float(row["importe"])
for row in csv.DictReader(f)
)
# Expresión generadora vs función generadora — cuándo usar cada una:
# Expresión: lógica simple, una línea, uso inmediato
suma = sum(x**2 for x in range(100))
# Función con yield: lógica compleja, múltiples yields, estado entre pasos
def procesar_lotes(datos, tamaño_lote):
lote = []
for item in datos:
lote.append(item)
if len(lote) == tamaño_lote:
yield lote
lote = []
if lote:
yield lote # último lote (posiblemente incompleto)
🔗 yield from: delegar en otro generador
# yield from iterable ≈ for x in iterable: yield x
# Pero también conecta send() y throw() al subgenerador
# Sin yield from — verboso
def aplanar_manual(listas):
for sublista in listas:
for item in sublista:
yield item
# Con yield from — idiomático
def aplanar(listas):
for sublista in listas:
yield from sublista
list(aplanar([[1, 2], [3, 4], [5, 6]])) # [1, 2, 3, 4, 5, 6]
# yield from acepta cualquier iterable
def cadena(*iterables):
for it in iterables:
yield from it
list(cadena([1, 2], "ab", range(3))) # [1, 2, 'a', 'b', 0, 1, 2]
# Recursión con yield from — aplanar árbol de profundidad arbitraria
def aplanar_arbol(nodo):
if isinstance(nodo, list):
for hijo in nodo:
yield from aplanar_arbol(hijo)
else:
yield nodo
arbol = [1, [2, [3, 4], 5], [6, 7]]
list(aplanar_arbol(arbol)) # [1, 2, 3, 4, 5, 6, 7]
# yield from con valor de retorno del subgenerador
def subgenerador():
yield 1
yield 2
return "resultado_sub" # este valor…
def generador_delegador():
valor = yield from subgenerador() # …llega aquí en 'valor'
print(f"Subgenerador devolvió: {valor}")
yield 3
gen = generador_delegador()
list(gen) # [1, 2, 3] y "Subgenerador devolvió: resultado_sub"
# Pipeline compuesto con yield from
def leer_fichero(ruta):
with open(ruta) as f:
yield from f
def filtrar_errores(lineas):
for linea in lineas:
if "ERROR" in linea:
yield linea.strip()
def parsear(lineas):
for linea in lineas:
partes = linea.split(" ", 3)
if len(partes) >= 4:
yield {"fecha": partes[0], "hora": partes[1],
"nivel": partes[2], "mensaje": partes[3]}
# Composición elegante — cada función solo sabe de su paso
def pipeline_logs(ruta):
yield from parsear(filtrar_errores(leer_fichero(ruta)))
📡 send() y throw(): comunicación bidireccional
# yield puede devolver un valor — convirtiendo el generador en una coroutine
def acumulador():
"""Coroutine que acumula valores recibidos vía send()."""
total = 0
while True:
valor = yield total # yield produce 'total' y espera recibir 'valor'
if valor is None:
break
total += valor
acc = acumulador()
next(acc) # 0 ← primera llamada: avanza hasta el primer yield
acc.send(10) # 10 ← envía 10, total=10, yield devuelve 10
acc.send(25) # 35 ← total=35
acc.send(5) # 40 ← total=40
acc.send(None) # StopIteration ← termina el bucle
# Decorador para "cebar" automáticamente una coroutine
from functools import wraps
def cebar(func):
@wraps(func)
def wrapper(*args, **kwargs):
gen = func(*args, **kwargs)
next(gen) # avanza hasta el primer yield
return gen
return wrapper
@cebar
def filtro_coroutine(criterio):
"""Coroutine que filtra y reenvía valores a otro destino."""
while True:
valor = yield
if criterio(valor):
print(f" Pasa el filtro: {valor}")
fc = filtro_coroutine(lambda x: x > 5)
fc.send(3) # (no imprime)
fc.send(8) # Pasa el filtro: 8
fc.send(2) # (no imprime)
fc.send(12) # Pasa el filtro: 12
# throw() — inyectar una excepción en el generador
def generador_robusto():
try:
while True:
valor = yield
print(f"Recibido: {valor}")
except ValueError as e:
print(f"Error controlado: {e}")
yield "recuperado" # puede seguir después del error
gen = generador_robusto()
next(gen)
gen.send(42) # Recibido: 42
gen.throw(ValueError, "dato inválido") # Error controlado: dato inválido → 'recuperado'
# close() — terminar el generador limpiamente (lanza GeneratorExit)
def generador_con_limpieza():
try:
while True:
yield
except GeneratorExit:
print("Limpiando recursos antes de cerrar") # ideal para cerrar ficheros, BD…
gen = generador_con_limpieza()
next(gen)
gen.close() # "Limpiando recursos antes de cerrar"
🧰 itertools: la caja de herramientas de iteración
import itertools
# ── ITERADORES INFINITOS ────────────────────────────────────────────────────
# count(inicio, paso) — cuenta desde inicio sin parar
gen = itertools.count(10, 2)
[next(gen) for _ in range(5)] # [10, 12, 14, 16, 18]
# cycle(iterable) — repite el iterable indefinidamente
colores = itertools.cycle(["rojo", "verde", "azul"])
[next(colores) for _ in range(7)] # ['rojo', 'verde', 'azul', 'rojo', 'verde', 'azul', 'rojo']
# repeat(valor, n) — repite un valor n veces (o infinitamente)
list(itertools.repeat("ok", 4)) # ['ok', 'ok', 'ok', 'ok']
list(map(pow, range(5), itertools.repeat(2))) # [0, 1, 4, 9, 16] ← x² para cada x
# ── COMBINATORIOS ───────────────────────────────────────────────────────────
list(itertools.permutations("ABC", 2))
# [('A','B'),('A','C'),('B','A'),('B','C'),('C','A'),('C','B')]
list(itertools.combinations("ABC", 2))
# [('A','B'),('A','C'),('B','C')] ← sin repetición, sin orden
list(itertools.combinations_with_replacement("AB", 2))
# [('A','A'),('A','B'),('B','B')] ← permite repetición
list(itertools.product("AB", repeat=2))
# [('A','A'),('A','B'),('B','A'),('B','B')] ← producto cartesiano
# ── TERMINADORES ────────────────────────────────────────────────────────────
# chain() — encadenar iterables sin crear listas intermedias
list(itertools.chain([1,2], [3,4], [5,6])) # [1,2,3,4,5,6]
list(itertools.chain.from_iterable([[1,2],[3,4],[5,6]])) # igual pero desde una lista de listas
# islice() — slice de un iterador (sin acceso aleatorio)
gen = (x**2 for x in range(1000))
list(itertools.islice(gen, 5)) # [0, 1, 4, 9, 16]
list(itertools.islice(gen, 2, 7)) # islice(iter, start, stop, step)
# takewhile / dropwhile — tomar/saltar mientras condición sea True
list(itertools.takewhile(lambda x: x < 5, [1, 3, 4, 6, 2, 8])) # [1, 3, 4]
list(itertools.dropwhile(lambda x: x < 5, [1, 3, 4, 6, 2, 8])) # [6, 2, 8]
# compress() — filtrar con una máscara de booleanos
datos = ["Ana", "Luis", "Marta", "Pedro"]
mascara = [True, False, True, False]
list(itertools.compress(datos, mascara)) # ['Ana', 'Marta']
# groupby() — agrupar elementos consecutivos por clave
datos = [
{"ciudad": "Madrid", "temp": 22},
{"ciudad": "Madrid", "temp": 25},
{"ciudad": "Barcelona", "temp": 20},
{"ciudad": "Barcelona", "temp": 18},
{"ciudad": "Sevilla", "temp": 30},
]
# IMPORTANTE: groupby requiere datos ORDENADOS por la clave
datos_ord = sorted(datos, key=lambda x: x["ciudad"])
for ciudad, grupo in itertools.groupby(datos_ord, key=lambda x: x["ciudad"]):
temps = [r["temp"] for r in grupo]
print(f"{ciudad}: media={sum(temps)/len(temps):.1f}°C")
# accumulate() — acumulado (como reduce pero devuelve todos los intermedios)
list(itertools.accumulate([1, 2, 3, 4, 5])) # [1, 3, 6, 10, 15]
list(itertools.accumulate([1, 2, 3, 4, 5], lambda a, b: a*b)) # [1, 2, 6, 24, 120]
import operator
list(itertools.accumulate([3, 1, 4, 1, 5], max)) # [3, 3, 4, 4, 5] ← max acumulado
# pairwise() (Python 3.10+) — pares consecutivos
list(itertools.pairwise([1, 2, 3, 4, 5])) # [(1,2),(2,3),(3,4),(4,5)]
# batched() (Python 3.12+) — dividir en lotes de tamaño n
list(itertools.batched(range(10), 3)) # [(0,1,2),(3,4,5),(6,7,8),(9,)]
📊 Memoria y rendimiento: cuándo usar generadores
import sys
import time
# ── COMPARATIVA DE MEMORIA ───────────────────────────────────────────────────
N = 1_000_000
lista_cuadrados = [x**2 for x in range(N)]
gen_cuadrados = (x**2 for x in range(N))
sys.getsizeof(lista_cuadrados) # ~8.000.056 bytes ≈ 8 MB
sys.getsizeof(gen_cuadrados) # ~208 bytes ← constante siempre
# ── COMPARATIVA DE TIEMPO (primera vez) ─────────────────────────────────────
# Crear lista: construye todos los valores → más lento de crear
t0 = time.perf_counter()
sum([x**2 for x in range(N)])
print(f"Lista: {time.perf_counter()-t0:.3f}s")
# Generador: produce sobre la marcha → ligeramente más rápido en total
t0 = time.perf_counter()
sum(x**2 for x in range(N))
print(f"Generador: {time.perf_counter()-t0:.3f}s")
# ── CUÁNDO USAR LISTA VS GENERADOR ───────────────────────────────────────────
# ✅ USA GENERADOR cuando:
# - Solo necesitas recorrerlo una vez
# - La secuencia es grande (>10.000 elementos) o infinita
# - Los elementos se calculan de forma costosa y puede que no uses todos
# - Compones varios pasos en un pipeline (evitas listas intermedias)
# - Procesas ficheros línea a línea
# ✅ USA LISTA cuando:
# - Necesitas acceso por índice: datos[42]
# - Necesitas len() antes de recorrer
# - Vas a recorrerlo varias veces
# - Necesitas reversed() o slicing arbitrario
# - Vas a pasarlo a código que espera una lista
# ── PIPELINE SIN LISTAS INTERMEDIAS ─────────────────────────────────────────
# Procesar un millón de registros sin construir ninguna lista intermedia
def leer_registros(n):
for i in range(n):
yield {"id": i, "valor": i * 1.5, "activo": i % 3 != 0}
def filtrar_activos(registros):
for r in registros:
if r["activo"]:
yield r
def transformar(registros):
for r in registros:
yield {**r, "valor_iva": round(r["valor"] * 1.21, 2)}
def top_n(registros, n=10):
import heapq
return heapq.nlargest(n, registros, key=lambda r: r["valor_iva"])
# Cada paso es lazy — en ningún momento se almacena el millón de registros
resultado = top_n(
transformar(
filtrar_activos(
leer_registros(1_000_000)
)
),
n=5
)
print(resultado)
# Pico de memoria: solo los 10 elementos del heap + el registro activo en curso
🛠️ Programa completo: pipeline de procesamiento de logs
"""
Pipeline de procesamiento de logs del servidor.
Analiza líneas de log en streaming (sin cargar el fichero completo),
filtra por nivel y rango de fechas, agrupa por hora, y genera
un informe de anomalías. Cada etapa es un generador.
"""
import re
import itertools
from datetime import datetime
from collections import defaultdict
# ── DATOS DE PRUEBA (simulamos un fichero de log) ────────────────────────────
LOG_SIMULADO = """
2026-03-10 08:01:22 INFO Servidor iniciado en puerto 8080
2026-03-10 08:01:45 INFO Base de datos conectada
2026-03-10 08:15:33 WARNING Tiempo de respuesta alto: 1.8s en /api/usuarios
2026-03-10 08:22:11 ERROR Fallo al conectar con servicio externo: timeout
2026-03-10 08:22:14 ERROR Reintento 1/3 fallido
2026-03-10 08:22:17 ERROR Reintento 2/3 fallido
2026-03-10 08:22:20 INFO Conexión restablecida
2026-03-10 09:05:01 INFO Tarea programada completada: 4821 registros
2026-03-10 09:31:44 WARNING Uso de memoria al 78%
2026-03-10 09:31:55 WARNING Uso de memoria al 82%
2026-03-10 09:32:10 ERROR OOM: proceso worker terminado
2026-03-10 09:32:15 INFO Worker reiniciado automáticamente
2026-03-10 10:00:00 INFO Backup nocturno iniciado
2026-03-10 10:45:22 ERROR Timeout en backup: tabla ventas_historico
2026-03-10 10:45:30 WARNING Backup parcial completado (87% de tablas)
2026-03-10 11:00:01 INFO Sistema estable — 238 peticiones/min
""".strip().splitlines()
# ── ETAPA 1: PARSEAR ─────────────────────────────────────────────────────────
PATRON_LOG = re.compile(
r"(?P\d{4}-\d{2}-\d{2})\s+"
r"(?P\d{2}:\d{2}:\d{2})\s+"
r"(?P\w+)\s+"
r"(?P.+)"
)
def parsear_logs(lineas):
for linea in lineas:
linea = linea.strip()
if not linea:
continue
m = PATRON_LOG.match(linea)
if m:
d = m.groupdict()
yield {
"ts": datetime.strptime(f"{d['fecha']} {d['hora']}", "%Y-%m-%d %H:%M:%S"),
"fecha": d["fecha"],
"hora": d["hora"],
"nivel": d["nivel"],
"mensaje": d["mensaje"],
}
# ── ETAPA 2: FILTRAR POR NIVEL ───────────────────────────────────────────────
def filtrar_nivel(registros, niveles=("WARNING", "ERROR")):
for r in registros:
if r["nivel"] in niveles:
yield r
# ── ETAPA 3: ENRIQUECER ──────────────────────────────────────────────────────
def enriquecer(registros):
for r in registros:
yield {
**r,
"hora_bloque": r["ts"].strftime("%H:00"), # agrupar por hora
"es_error": r["nivel"] == "ERROR",
}
# ── ETAPA 4: GENERAR ALERTAS (ventana deslizante) ────────────────────────────
def detectar_rafagas(registros, ventana_seg=60, umbral=3):
"""Alerta si hay >= umbral errores en una ventana de tiempo."""
buffer = []
for r in registros:
yield r # siempre deja pasar el registro
if r["es_error"]:
buffer.append(r["ts"])
# Limpiar buffer de errores fuera de la ventana
buffer = [t for t in buffer if (r["ts"] - t).seconds <= ventana_seg]
if len(buffer) == umbral: # exactamente al llegar al umbral
yield {
**r,
"nivel": "ALERT",
"mensaje": f"🚨 RÁFAGA: {umbral} errores en {ventana_seg}s",
"es_error": False,
}
# ── ANÁLISIS: AGRUPAR POR HORA ───────────────────────────────────────────────
def agrupar_por_hora(registros):
"""Agrupa registros por hora y cuenta por nivel."""
resumen = defaultdict(lambda: defaultdict(int))
todos = list(registros) # materializar para poder iterar dos veces
for r in todos:
resumen[r["hora_bloque"]][r["nivel"]] += 1
return resumen, todos
# ── EJECUCIÓN DEL PIPELINE ───────────────────────────────────────────────────
pipeline = detectar_rafagas(
enriquecer(
filtrar_nivel(
parsear_logs(LOG_SIMULADO),
niveles=("WARNING", "ERROR")
)
)
)
resumen_horas, registros_finales = agrupar_por_hora(pipeline)
# ── INFORME ───────────────────────────────────────────────────────────────────
print(f"\n{'═'*58}")
print(f"{'INFORME DE ANOMALÍAS DEL SERVIDOR':^58}")
print(f"{'═'*58}")
print(f"\n {'HORA':<10} {'WARNING':>8} {'ERROR':>8} {'ALERT':>8}")
print(f" {'-'*38}")
for hora in sorted(resumen_horas):
w = resumen_horas[hora].get("WARNING", 0)
e = resumen_horas[hora].get("ERROR", 0)
a = resumen_horas[hora].get("ALERT", 0)
alerta_str = " ⚠️" if a > 0 else ""
print(f" {hora:<10} {w:>8} {e:>8} {a:>8}{alerta_str}")
print(f"\n {'─'*58}")
total_w = sum(v.get("WARNING", 0) for v in resumen_horas.values())
total_e = sum(v.get("ERROR", 0) for v in resumen_horas.values())
total_a = sum(v.get("ALERT", 0) for v in resumen_horas.values())
print(f" {'TOTAL':<10} {total_w:>8} {total_e:>8} {total_a:>8}")
print(f"\n DETALLE DE EVENTOS:")
print(f" {'─'*58}")
for r in registros_finales:
icono = {"WARNING": "⚠️ ", "ERROR": "❌", "ALERT": "🚨"}.get(r["nivel"], " ")
print(f" {icono} {r['hora']} {r['nivel']:<8} {r['mensaje'][:45]}")
print(f"\n{'═'*58}\n")
# ── BONUS: itertools en el pipeline ──────────────────────────────────────────
# Agrupar con itertools.groupby
errores_solo = [r for r in registros_finales if r["nivel"] == "ERROR"]
errores_ord = sorted(errores_solo, key=lambda r: r["hora_bloque"])
print(" Errores agrupados por hora (itertools.groupby):")
for hora, grupo in itertools.groupby(errores_ord, key=lambda r: r["hora_bloque"]):
mensajes = [r["mensaje"][:35] for r in grupo]
print(f" {hora}: {len(mensajes)} error(es)")
🐛 Errores clásicos con generadores
1. Intentar reutilizar un generador agotado
gen = (x**2 for x in range(5))
list(gen) # [0, 1, 4, 9, 16]
list(gen) # [] ← agotado — bug silencioso
# ✅ Guardar en lista si necesitas reutilizar
datos = list(x**2 for x in range(5))
sum(datos) # ✅
max(datos) # ✅
2. Llamar a la función generadora sin consumir el resultado
def mis_numeros():
yield 1; yield 2; yield 3
mis_numeros() # ← crea el generador pero lo descarta inmediatamente
print(mis_numeros()) # ← no es una lista
# ✅ Consumir el generador
list(mis_numeros()) # [1, 2, 3]
for n in mis_numeros(): print(n)
3. send() sin cebar primero el generador
def mi_coroutine():
valor = yield
print(f"Recibido: {valor}")
gen = mi_coroutine()
gen.send(42) # TypeError: can't send non-None value to a just-started generator
# ✅ Siempre next() o send(None) primero
gen = mi_coroutine()
next(gen) # avanza hasta el primer yield
gen.send(42) # ahora sí: "Recibido: 42"
4. Modificar la colección que estás iterando
lista = [1, 2, 3, 4, 5]
for x in lista:
if x % 2 == 0:
lista.remove(x) # ❌ comportamiento indefinido: salta elementos
# ✅ Iterar sobre una copia, o usar comprensión
lista = [x for x in lista if x % 2 != 0] # [1, 3, 5]
5. groupby() sin ordenar antes
datos = [("A", 1), ("B", 2), ("A", 3), ("B", 4)]
# ❌ groupby solo agrupa consecutivos — "A" aparece dos veces
for k, g in itertools.groupby(datos, key=lambda x: x[0]):
print(k, list(g))
# A [('A', 1)] ← solo el primero
# B [('B', 2)]
# A [('A', 3)] ← "A" de nuevo
# B [('B', 4)]
# ✅ Ordenar primero
datos_ord = sorted(datos, key=lambda x: x[0])
for k, g in itertools.groupby(datos_ord, key=lambda x: x[0]):
print(k, list(g))
# A [('A', 1), ('A', 3)]
# B [('B', 2), ('B', 4)]
✅ Resumen y próximos pasos
Los generadores son la herramienta de Python para la evaluación perezosa: permiten trabajar con secuencias de cualquier tamaño con memoria constante, componer pipelines de transformación sin listas intermedias y, con send(), construir coroutines bidireccionales. itertools completa el arsenal con combinatorios, agrupación, acumulados y encadenamiento que eliminan la necesidad de bucles manuales en la mayoría de casos.
Con esta lección se cierra el Módulo 4 — Funciones y módulos. El siguiente módulo, ya completado, aborda la Programación Orientada a Objetos. Solo queda una lección pendiente en todo el curso: Introducción a Machine Learning, que cerrará el Módulo 8.
❓ Preguntas frecuentes
❓ Preguntas frecuentes sobre Generadores e iteradores en Python: yield, lazy evaluation y memoria eficiente
Las dudas más comunes respondidas de forma clara y directa.
Un iterable es cualquier objeto que puede devolver un iterador: listas, tuplas, strings, dicts, sets, ficheros, range()... Implementa __iter__(). Un iterador es un objeto que recuerda su posición actual y sabe cómo obtener el siguiente elemento: implementa tanto __iter__() (devuelve self) como __next__() (devuelve el siguiente valor o lanza StopIteration). Un generador es un tipo especial de iterador creado con una función que contiene yield, o con una expresión generadora. Todo generador es un iterador, pero no todo iterador es un generador. La diferencia práctica: los generadores se crean con una sintaxis mucho más concisa y gestionan el estado automáticamente.
Porque un generador es un iterador con estado interno: recuerda en qué punto del código se quedó tras el último yield. Una vez que StopIteration se lanza, el estado interno del generador queda marcado como agotado y no hay forma de "rebobinar". Si necesitas recorrer los valores varias veces, tienes tres opciones: (1) guardar el resultado en una lista: datos = list(mi_generador()), (2) crear una nueva instancia del generador llamando de nuevo a la función generadora, o (3) usar itertools.tee() para clonar el generador en dos iteradores independientes, aunque esto consume memoria si los dos iteradores van a ritmos muy distintos.
Usa la expresión generadora cuando no necesites todos los valores a la vez, cuando la secuencia sea muy larga o potencialmente infinita, o cuando solo vayas a recorrerla una vez. La comprensión de lista construye toda la lista en memoria de golpe: si tienes 10 millones de elementos, ocupa esa memoria completa. La expresión generadora produce cada elemento bajo demanda: ocupa memoria constante independientemente del tamaño. La regla práctica: si lo pasas directamente a sum(), max(), any(), all(), join() o un bucle for, usa expresión generadora. Si necesitas indexar, slicing, len() o reutilizarlo, usa lista.
yield from iterable es equivalente a for x in iterable: yield x, pero más eficiente y con una diferencia crucial: también conecta send() y throw() del generador externo directamente al interno, lo que permite crear coroutines correctamente. Se usa principalmente en tres casos: (1) delegar en otro generador para aplanar código o dividir responsabilidades, (2) aplanar secuencias anidadas de forma recursiva, y (3) como base de coroutines con asyncio (aunque asyncio usa async/await, que internamente funciona sobre yield from). Es la forma idiomática de componer generadores.
next(gen) es equivalente a gen.send(None): reanuda el generador hasta el siguiente yield y devuelve el valor que yield produce. gen.send(valor) hace lo mismo, pero además inyecta valor dentro del generador: la expresión x = yield devuelve ese valor en x. Esto convierte al generador en una coroutine bidireccional: puede producir valores hacia afuera con yield y recibir valores desde afuera con send(). La limitación: la primera llamada siempre debe ser next(gen) o gen.send(None) porque el generador empieza antes del primer yield y no hay ningún yield esperando recibir un valor todavía.
★
★
★
★
★
Valora este artículo
¿Útil?
💬 Foro de discusión
¿Tienes dudas sobre Generadores e iteradores en Python: yield, lazy evaluation y memoria eficiente? Comparte tu pregunta con la comunidad.
¿Tienes cuenta?
o comenta como invitado ↓
Todavía no hay mensajes. ¡Sé el primero en participar!