pipes & filter example

pipes_and_filter.py

import time
from functools import wraps

# 1. Definición de los interceptores


def interceptor_de_cache(cache):
    """Interceptor que verifica si el resultado ya está en cache."""

    def wrapper(datos):
        key = str(datos)  # Generamos una clave simple para el cache
        if key in cache:
            print("Cache hit!")
            return cache[key]
        return datos

    return wrapper


def interceptor_de_rendimiento(func):
    """Interceptor para medir el tiempo de ejecución."""

    @wraps(func)
    def wrapper(datos):
        inicio = time.time()
        resultado = func(datos)
        fin = time.time()
        print(f"Tiempo de procesamiento: {fin - inicio:.4f} segundos")
        return resultado

    return wrapper


# 2. Clase base para los filtros


class Filtro:
    def __init__(self):
        self.interceptores = []

    def add_interceptor(self, interceptor):
        """Agregar interceptores que se ejecutarán en cada etapa."""
        self.interceptores.append(interceptor)

    def _apply_interceptors(self, datos):
        """Aplicar los interceptores antes de procesar los datos."""
        for interceptor in self.interceptores:
            datos = interceptor(datos)
        return datos

    def process(self, datos):
        """Método principal para procesar los datos."""
        datos = self._apply_interceptors(datos)  # Llamar interceptores
        return self._procesar(datos)

    def _procesar(self, datos):
        """Método abstracto que debe ser implementado en cada filtro."""
        raise NotImplementedError("Debe implementarse el procesamiento específico en cada filtro")

    def before_process(self, datos):
        """Hook que se ejecuta antes de procesar los datos."""
        return datos

    def after_process(self, datos):
        """Hook que se ejecuta después de procesar los datos."""
        return datos


# 3. Filtro específico que usa interceptores (FiltroMayusculas)


class FiltroMayusculas(Filtro):
    def __init__(self):
        super().__init__()
        self.cache = {}
        self.add_interceptor(interceptor_de_cache(self.cache))  # Agregamos el interceptor de cache
        self.add_interceptor(
            interceptor_de_rendimiento(self._procesar)
        )  # Agregamos el interceptor de rendimiento

    def _procesar(self, datos):
        return [dato.upper() for dato in datos]


# 4. Otro filtro: Inversor de cadenas (FiltroReversa)


class FiltroReversa(Filtro):
    def __init__(self):
        super().__init__()

    def _procesar(self, datos):
        return [dato[::-1] for dato in datos]


# 5. La clase Pipeline que orquesta los filtros


class Pipeline:
    def __init__(self, filtros):
        self.filtros = filtros

    def ejecutar(self, datos):
        for filtro in self.filtros:
            datos = filtro.before_process(datos)  # Ejecutamos el hook antes
            datos = filtro.process(datos)  # Procesamos los datos
            datos = filtro.after_process(datos)  # Ejecutamos el hook después
        return datos


# 6. Uso del Pipeline

# Creamos los filtros
filtro_mayusculas = FiltroMayusculas()
filtro_reversa = FiltroReversa()

# Creamos el pipeline con estos filtros
pipeline = Pipeline([filtro_mayusculas, filtro_reversa])

# Datos de entrada
datos = ["hola", "mundo", "python"]

# Ejecutamos el pipeline
resultado = pipeline.ejecutar(datos)

# Imprimimos el resultado final
print("Resultado final:", resultado)