Los Decoradores y Generadores son características avanzadas de Python que separan a los programadores intermedios de los verdaderos profesionales. Los decoradores te permiten modificar de forma elegante el comportamiento de las funciones, mientras que los generadores crean iteradores altamente eficientes en memoria. En esta guía completa, dominarás ambos conceptos con ejemplos prácticos del mundo real.
🎨 ¿Qué Son los Decoradores?
Los decoradores son esencialmente funciones que modifican el comportamiento de otras funciones sin alterar realmente su código fuente. Piensa en ellos como envoltorios que añaden funcionalidad extra a una función existente.
# Una función decoradora simple
def mi_decorador(funcion):
def envoltura():
print("🚀 Antes de la ejecución de la función")
funcion()
print("✅ Después de la ejecución de la función")
return envoltura
@mi_decorador
def saludo():
print("¡Hola, Universo Python!")
# Llamando a la función decorada
saludo()
# Salida:
# 🚀 Antes de la ejecución de la función
# ¡Hola, Universo Python!
# ✅ Después de la ejecución de la funciónEl símbolo @ es simplemente "azúcar sintáctico" para la siguiente operación:
# Esta sintaxis:
@mi_decorador
def saludo():
print("¡Hola!")
# Es exactamente equivalente a:
def saludo():
print("¡Hola!")
saludo = mi_decorador(saludo)📝 Creando Decoradores Prácticos
Decorador con Argumentos de Función
Si la función que estás decorando acepta argumentos, tu envoltura también debe aceptarlos. Usamos *args y **kwargs para esto.
def registrar_llamada(funcion):
"""Registra exactamente cuándo se llama a una función y sus argumentos"""
def envoltura(*args, **kwargs):
print(f"📞 Llamando función: {funcion.__name__}")
print(f" Args Posicionales: {args}")
print(f" Kwargs: {kwargs}")
resultado = funcion(*args, **kwargs)
print(f" Retornó: {resultado}")
return resultado
return envoltura
@registrar_llamada
def sumar_numeros(a, b):
return a + b
@registrar_llamada
def saludo_personalizado(nombre, texto_saludo="Hola"):
return f"{texto_saludo}, {nombre}!"
print(sumar_numeros(10, 5))
print(saludo_personalizado("Python", texto_saludo="Bienvenido"))Preservando Metadatos con functools.wraps
Cuando decoras una función, pierde sus metadatos originales (como su nombre y docstring). Para solucionar esto, usa siempre @wraps de la biblioteca integrada functools.
from functools import wraps
def decorador_profesional(funcion):
@wraps(funcion) # Preserva __name__, __doc__, etc.
def envoltura(*args, **kwargs):
"""Documentación interna de la envoltura"""
return funcion(*args, **kwargs)
return envoltura
@decorador_profesional
def mi_funcion():
"""Documentación original de la función"""
pass
print(mi_funcion.__name__) # Imprime 'mi_funcion' (no 'envoltura')
print(mi_funcion.__doc__) # Imprime 'Documentación original de la función'Decoradores que Aceptan Parámetros
from functools import wraps
def repetir_ejecucion(veces):
"""Decorador que repite la función N veces"""
def decorador(funcion):
@wraps(funcion)
def envoltura(*args, **kwargs):
resultados = []
for _ in range(veces):
resultado = funcion(*args, **kwargs)
resultados.append(resultado)
return resultados
return envoltura
return decorador
@repetir_ejecucion(veces=3)
def decir_hola():
print("¡Hola!")
return "dijo hola"
resultados = decir_hola()
# Salida: ¡Hola! (impreso 3 veces)
print(resultados) # ['dijo hola', 'dijo hola', 'dijo hola']⚡ Decoradores Prácticos Muy Útiles
Temporizador: Midiendo el Tiempo de Ejecución
import time
from functools import wraps
def temporizador(funcion):
"""Mide el tiempo exacto de ejecución de una función"""
@wraps(funcion)
def envoltura(*args, **kwargs):
tiempo_inicio = time.perf_counter()
resultado = funcion(*args, **kwargs)
tiempo_fin = time.perf_counter()
duracion = tiempo_fin - tiempo_inicio
print(f"⏱️ {funcion.__name__} se ejecutó en {duracion:.4f} segundos")
return resultado
return envoltura
@temporizador
def procesar_datos():
time.sleep(1) # Simulando procesamiento pesado de datos
return "Datos procesados con éxito"
procesar_datos() # ⏱️ procesar_datos se ejecutó en 1.0012 segundosCaché y Memorización
Si tienes una función costosa, puedes almacenar en caché sus resultados para evitar repetir cálculos.
from functools import wraps
def cache_simple(funcion):
"""Un caché simple basado en diccionario para resultados de funciones"""
memoria = {}
@wraps(funcion)
def envoltura(*args):
if args in memoria:
print(f"📦 Acierto de caché para {args}")
return memoria[args]
resultado = funcion(*args)
memoria[args] = resultado
print(f"💾 Guardando nuevo resultado en caché: {args}")
return resultado
return envoltura
@cache_simple
def fibonacci_recursivo(n):
"""Calcula la secuencia de Fibonacci recursivamente"""
if n < 2:
return n
return fibonacci_recursivo(n-1) + fibonacci_recursivo(n-2)
print(fibonacci_recursivo(10)) # ¡Extremadamente rápido gracias al caché!
# Nota: Python 3.9+ incluye lru_cache incorporado que está muy optimizado:
from functools import lru_cache
@lru_cache(maxsize=128)
def fibonacci_optimizado(n):
if n < 2:
return n
return fibonacci_optimizado(n-1) + fibonacci_optimizado(n-2)Mecanismo de Reintento (Retry)
import time
from functools import wraps
def reintentar_si_falla(intentos=3, retraso=1, excepciones=(Exception,)):
"""Reintenta ejecutar una función varias veces si falla"""
def decorador(funcion):
@wraps(funcion)
def envoltura(*args, **kwargs):
ultima_excepcion = None
for intento in range(1, intentos + 1):
try:
return funcion(*args, **kwargs)
except excepciones as e:
ultima_excepcion = e
print(f"⚠️ Intento {intento}/{intentos} falló: {e}")
if intento < intentos:
print(f" Esperando {retraso} segundos antes de reintentar...")
time.sleep(retraso)
raise ultima_excepcion
return envoltura
return decorador
@reintentar_si_falla(intentos=3, retraso=1)
def obtener_datos_api():
import random
if random.random() < 0.7: # 70% de probabilidad de falla
raise ConnectionError("Falla en la conexión de red")
return {"status": "success"}
try:
resultado = obtener_datos_api()
print(f"✅ Éxito: {resultado}")
except ConnectionError as e:
print(f"❌ Falló permanentemente después de todos los intentos: {e}")Autenticación y Autorización
from functools import wraps
# Simulando una sesión de usuario que ha iniciado sesión
usuario_actual = {"nombre": "Admin", "rol": "admin"}
def requiere_inicio_sesion(funcion):
"""Verifica si el usuario ha iniciado sesión actualmente"""
@wraps(funcion)
def envoltura(*args, **kwargs):
if not usuario_actual:
raise PermissionError("❌ ¡Por favor, inicia sesión primero!")
return funcion(*args, **kwargs)
return envoltura
def requiere_rol(*roles_permitidos):
"""Verifica si el usuario conectado tiene el permiso requerido"""
def decorador(funcion):
@wraps(funcion)
def envoltura(*args, **kwargs):
if usuario_actual.get("rol") not in roles_permitidos:
raise PermissionError(
f"❌ Acceso denegado. Roles requeridos: {roles_permitidos}"
)
return funcion(*args, **kwargs)
return envoltura
return decorador
@requiere_inicio_sesion
@requiere_rol("admin", "moderador")
def eliminar_cuenta_usuario(id_usuario):
return f"✅ La cuenta de usuario {id_usuario} se eliminó correctamente"
print(eliminar_cuenta_usuario(42))🔄 ¿Qué Son los Generadores?
Los generadores son funciones especiales que producen una secuencia de valores bajo demanda, utilizando la palabra clave yield en lugar de return. Son increíblemente eficientes en cuanto a memoria porque solo generan un valor a la vez, en lugar de almacenar todo en la memoria simultáneamente.
# Función estándar: almacena todo en la memoria
def crear_lista(n):
resultado = []
for i in range(n):
resultado.append(i ** 2)
return resultado
# Función generadora: genera elementos bajo demanda
def crear_generador(n):
for i in range(n):
yield i ** 2
# Comparación de memoria
import sys
lista_estandar = crear_lista(1000000)
objeto_generador = crear_generador(1000000)
print(f"Memoria de la Lista: {sys.getsizeof(lista_estandar):,} bytes") # ~8MB
print(f"Memoria del Generador: {sys.getsizeof(objeto_generador)} bytes") # ¡~120 bytes!📝 Construyendo Generadores
Ejemplo de Generador Simple
def temporizador_cuenta_regresiva(n):
"""Una función generadora de cuenta regresiva"""
print("🚀 Iniciando secuencia de cuenta regresiva...")
while n > 0:
yield n
n -= 1
print("🔥 ¡Despegue!")
# Usándolo en un bucle for
for numero in temporizador_cuenta_regresiva(5):
print(numero)
# O avanzando manualmente usando next()
gen = temporizador_cuenta_regresiva(3)
print(next(gen)) # 3
print(next(gen)) # 2
print(next(gen)) # 1
# print(next(gen)) # Levanta la excepción StopIterationGeneradores Infinitos
def numeros_pares_infinitos():
"""Un generador infinito que produce números pares"""
n = 0
while True:
yield n
n += 2
# ¡Úsalo con precaución y condiciones de ruptura!
pares = numeros_pares_infinitos()
for _ in range(10):
print(next(pares), end=" ")
# Salida: 0 2 4 6 8 10 12 14 16 18Generadores con Múltiples Yields
def leer_archivo_grande_en_trozos(ruta_archivo, tamano_trozo=1024):
"""Lee un archivo masivo eficientemente en trozos"""
with open(ruta_archivo, 'r') as archivo:
while True:
trozo = archivo.read(tamano_trozo)
if not trozo:
break
yield trozo
# Puedes procesar un archivo de 10GB sin agotar tu memoria
# for trozo in leer_archivo_grande_en_trozos("archivo_log_masivo.txt"):
# procesar_datos(trozo)Esto es extremadamente útil cuando se trata de ciencia de datos. Aprende más sobre la manipulación de archivos en nuestra guía sobre el manejo de archivos en Python.
🎯 Expresiones Generadoras
Las expresiones generadoras son muy similares a las list comprehensions de Python, pero usan paréntesis en lugar de corchetes.
# List comprehension: crea la lista completa en la memoria
lista_cuadrados = [x**2 for x in range(1000000)]
# Expresión generadora: genera perezosamente bajo demanda
gen_cuadrados = (x**2 for x in range(1000000))
# Se puede usar directamente dentro de funciones sin esfuerzo
suma_total = sum(x**2 for x in range(1000)) # No se necesitan corchetes adicionales
promedio = sum(x for x in range(100)) / 100⚡ Delegación con Yield From
def generador_anidado():
yield from range(3) # Produce 0, 1, 2
yield from "abc" # Produce a, b, c
yield from [10, 20, 30] # Produce 10, 20, 30
for item in generador_anidado():
print(item, end=" ")
# Salida: 0 1 2 a b c 10 20 30
# Un ejemplo práctico brillante: aplanar listas profundamente anidadas
def aplanar_lista(lista_anidada):
"""Aplana recursivamente listas anidadas"""
for item in lista_anidada:
if isinstance(item, list):
yield from aplanar_lista(item)
else:
yield item
lista_compleja = [1, [2, 3, [4, 5]], 6, [7, 8]]
print(list(aplanar_lista(lista_compleja)))
# Salida: [1, 2, 3, 4, 5, 6, 7, 8]🎯 Proyecto Práctico: Pipeline Avanzado de Procesamiento de Datos
Construyamos un sofisticado sistema de procesamiento de datos utilizando decoradores y generadores, aplicando conceptos modernos de Programación Orientada a Objetos (POO).
import time
import random
from functools import wraps
from typing import Generator, Callable, List
from dataclasses import dataclass
# ============================================
# DECORADORES DEL SISTEMA
# ============================================
def temporizador_pipeline(funcion):
"""Mide el tiempo de ejecución del pipeline"""
@wraps(funcion)
def envoltura(*args, **kwargs):
inicio = time.perf_counter()
resultado = funcion(*args, **kwargs)
duracion = time.perf_counter() - inicio
print(f"⏱️ {funcion.__name__} terminó en: {duracion:.4f}s")
return resultado
return envoltura
def registrar_paso_pipeline(funcion):
"""Registra los pasos de ejecución del pipeline"""
@wraps(funcion)
def envoltura(*args, **kwargs):
print(f"📍 Iniciando Paso: {funcion.__name__}")
resultado = funcion(*args, **kwargs)
print(f"✅ Paso Completado: {funcion.__name__}")
return resultado
return envoltura
# ============================================
# GENERADORES DEL PIPELINE
# ============================================
@dataclass
class DatosCrudos:
id: int
valor: float
timestamp: float
def simular_flujo_datos(cantidad: int) -> Generator[DatosCrudos, None, None]:
"""Generador que simula una fuente de datos activa"""
for i in range(cantidad):
yield DatosCrudos(
id=i,
valor=random.uniform(0, 100),
timestamp=time.time()
)
time.sleep(0.01) # Simulando latencia de red
def filtrar_valores_bajos(flujo_datos: Generator, umbral_minimo: float = 20) -> Generator:
"""Filtra los datos por debajo del umbral mínimo"""
for datos in flujo_datos:
if datos.valor >= umbral_minimo:
yield datos
def transformar_a_diccionario(flujo_datos: Generator) -> Generator[dict, None, None]:
"""Transforma los datos del objeto en un formato de diccionario procesado"""
for datos in flujo_datos:
yield {
"id": datos.id,
"valor_original": datos.valor,
"valor_procesado": round(datos.valor * 1.1, 2),
"categoria": "Alto" if datos.valor > 50 else "Bajo",
"procesado_en": time.time()
}
def agrupar_datos(flujo_datos: Generator, tamano_lote: int = 10) -> Generator[List, None, None]:
"""Agrupa los datos procesados en tamaños de lote específicos"""
lote = []
for datos in flujo_datos:
lote.append(datos)
if len(lote) >= tamano_lote:
yield lote
lote = []
if lote: # Produciendo el lote parcial final
yield lote
# ============================================
# CLASE PRINCIPAL DEL PIPELINE
# ============================================
class PipelineDeDatos:
"""Sistema central de pipeline de procesamiento de datos"""
def __init__(self, nombre: str):
self.nombre = nombre
self.pasos: List[Callable] = []
self.estadisticas = {
"total_elementos_procesados": 0,
"total_lotes_generados": 0
}
def agregar_paso(self, funcion_paso: Callable) -> 'PipelineDeDatos':
"""Agrega un paso de procesamiento usando una interfaz fluida"""
self.pasos.append(funcion_paso)
return self
@temporizador_pipeline
@registrar_paso_pipeline
def ejecutar_pipeline(self, flujo_entrada: Generator) -> Generator:
"""Ejecuta todo el pipeline secuencialmente"""
flujo = flujo_entrada
for paso in self.pasos:
flujo = paso(flujo)
return flujo
def recolectar_resultados(self, flujo_procesado: Generator) -> List[dict]:
"""Consume el generador y recopila los resultados finales"""
resultados_finales = []
for lote in flujo_procesado:
self.estadisticas["total_lotes_generados"] += 1
for item in lote:
resultados_finales.append(item)
self.estadisticas["total_elementos_procesados"] += 1
return resultados_finales
def mostrar_reporte(self):
"""Muestra las estadísticas de ejecución del pipeline"""
print(f"\n📊 REPORTE: {self.nombre}")
print("=" * 40)
for clave, valor in self.estadisticas.items():
print(f" {clave}: {valor}")
# ============================================
# DEMOSTRACIÓN DE EJECUCIÓN
# ============================================
if __name__ == "__main__":
print("=" * 60)
print("🚀 SISTEMA AVANZADO DE PIPELINE DE DATOS")
print("=" * 60)
# Inicializar el pipeline
pipeline = PipelineDeDatos("Pipeline de Ventas Empresariales")
# Configurar los pasos de procesamiento
pipeline.agregar_paso(
lambda datos: filtrar_valores_bajos(datos, umbral_minimo=25)
).agregar_paso(
transformar_a_diccionario
).agregar_paso(
lambda datos: agrupar_datos(datos, tamano_lote=5)
)
# Generar el flujo de datos entrante
flujo_entrante = simular_flujo_datos(50)
# Ejecutar el pipeline (los generadores hacen que esto sea perezoso e instantáneo)
print("\n" + "-" * 60)
flujo_procesado = pipeline.ejecutar_pipeline(flujo_entrante)
# Recopilar los resultados (aquí es donde ocurre el cálculo real)
resultados_finales = pipeline.recolectar_resultados(flujo_procesado)
# Mostrar el informe final
pipeline.mostrar_reporte()
print("\n📄 Primeros 3 Resultados Procesados:")
for item in resultados_finales[:3]:
print(f" {item}")
print(f"\n✅ Total de elementos válidos procesados: {len(resultados_finales)}")Este proyecto es perfecto para agregarlo a tu portafolio de Python, ya que demuestra un profundo conocimiento de la gestión de la memoria y la arquitectura limpia.
💡 Cuándo Usar Cada Herramienta
Debes usar Decoradores específicamente para:
- ✅ Implementar sistemas de registro y monitoreo automáticamente.
- ✅ Crear capas de almacenamiento en caché y memorización.
- ✅ Manejar verificaciones de autenticación y autorización basadas en roles.
- ✅ Realizar una validación estricta de las entradas antes de la ejecución de la función.
- ✅ Agregar lógica de reintento de red sin abarrotar la lógica empresarial central.
- ✅ Medir el tiempo de ejecución con precisión.
Debes usar Generadores específicamente para:
- ✅ Procesar archivos extremadamente grandes que superan la RAM disponible.
- ✅ Manejar datos de transmisión en vivo de forma continua.
- ✅ Construir canales de procesamiento de datos complejos y eficientes en memoria.
- ✅ Representar secuencias matemáticas infinitas.
- ✅ Ahorrar cantidades significativas de memoria del sistema con elegancia.
Para obtener documentación técnica oficial con respecto a estas características, siempre revisa la documentación de functools en Python.
Conclusión
Dominar los Decoradores y Generadores transformará tu código Python de funcional a altamente profesional. Proporcionan soluciones elegantes a problemas arquitectónicos complejos y garantizan que tus aplicaciones sigan siendo altamente funcionales y fáciles de mantener.