lección 4

Validación con Python puro: funciones y assertions

Escribir validadores reutilizables, patterns comunes, cuándo usar assert vs raise, y decoradores de validación.

⏱ 50 min

### Antes de los frameworks: el poder del Python desnudo

Antes de instalar Great Expectations o cualquier framework de validación, necesitas dominar la validación con Python puro. ¿Por qué? Tres razones. Primera: entenderás QUÉ hace el framework por debajo, así que cuando falle sabrás depurarlo. Segunda: no siempre puedes (o necesitas) instalar un framework — a veces un script de 50 líneas es suficiente. Tercera: aprenderás patterns que usarás durante TODA tu carrera, con o sin frameworks.

Analogía: antes de usar una calculadora científica, aprendes aritmética a mano. No porque vayas a calcular raíces cuadradas de cabeza el resto de tu vida, sino porque necesitas ENTENDER qué hace la calculadora para saber cuándo el resultado no tiene sentido. Con la validación de datos es igual.

### assert vs raise: dos filosofías diferentes

Python tiene dos mecanismos principales para "gritar cuando algo está mal": assert y raise. Parecen iguales pero tienen una diferencia FUNDAMENTAL que muchos juniors confunden, y esa confusión puede causarte problemas en producción.

1# ASSERT: verificación durante DESARROLLO
2# Se desactiva con python -O (optimized mode)
3# NUNCA uses assert para validación en producción
4
5def procesar_lote_desarrollo(df):
6    # Esto es un "sanity check" para el desarrollador
7    assert len(df) > 0, "El DataFrame está vacío — ¿el pipeline upstream falló?"
8    assert 'importe' in df.columns, "Falta la columna 'importe' — ¿cambió el schema?"
9    # ... procesamiento ...
10
11# RAISE: verificación en PRODUCCIÓN
12# SIEMPRE se ejecuta, no se puede desactivar
13# Usa esto para validación real de datos
14
15def procesar_lote_produccion(df):
16    if len(df) == 0:
17        raise ValueError("El DataFrame está vacío — abortando pipeline")
18    if 'importe' not in df.columns:
19        raise KeyError("Falta columna 'importe' — posible cambio de schema upstream")
20    if df['importe'].isna().sum() > len(df) * 0.05:
21        raise ValueError(f"Más del 5% de importes son NULL ({df['importe'].isna().sum()}/{len(df)})")
22    # ... procesamiento ...

assert es para desarrollo (sanity checks). raise es para producción (validación real).

NUNCA uses assert para validación en producción. Python puede ejecutarse con la flag -O (optimized) que DESACTIVA todos los asserts. Si tu validación depende de asserts, un día alguien arrancará el proceso con -O y todas tus validaciones desaparecerán silenciosamente. Usa assert solo para sanity checks durante desarrollo. Para producción: raise con excepciones apropiadas.

### Pattern 1: Funciones validadoras puras

El pattern más simple y poderoso: funciones que reciben datos y retornan un resultado de validación. No modifican nada, no tienen efectos secundarios, solo inspeccionan y reportan. Son fáciles de testear con pytest, fáciles de combinar y fáciles de entender.

1from dataclasses import dataclass, field
2from typing import Callable
3import pandas as pd
4
5@dataclass
6class ResultadoValidacion:
7    """Resultado de una validación individual."""
8    nombre: str
9    paso: bool
10    mensaje: str
11    registros_afectados: int = 0
12    severidad: str = 'ERROR'  # ERROR, WARNING, INFO
13
14@dataclass 
15class InformeValidacion:
16    """Informe completo de todas las validaciones."""
17    resultados: list[ResultadoValidacion] = field(default_factory=list)
18    
19    @property
20    def paso_todo(self) -> bool:
21        return all(r.paso for r in self.resultados if r.severidad == 'ERROR')
22    
23    @property
24    def errores(self) -> list[ResultadoValidacion]:
25        return [r for r in self.resultados if not r.paso and r.severidad == 'ERROR']
26    
27    @property
28    def warnings(self) -> list[ResultadoValidacion]:
29        return [r for r in self.resultados if not r.paso and r.severidad == 'WARNING']
30    
31    def resumen(self) -> str:
32        total = len(self.resultados)
33        pasaron = sum(1 for r in self.resultados if r.paso)
34        return f"{pasaron}/{total} checks pasaron | {len(self.errores)} errores | {len(self.warnings)} warnings"
35
36
37def check_no_nulos(df: pd.DataFrame, columna: str, max_pct: float = 0.0) -> ResultadoValidacion:
38    """Verifica que una columna no tenga más nulos que el umbral."""
39    pct_nulos = df[columna].isna().mean() * 100
40    paso = pct_nulos <= max_pct
41    return ResultadoValidacion(
42        nombre=f"no_nulos_{columna}",
43        paso=paso,
44        mensaje=f"{columna}: {pct_nulos:.1f}% nulos (umbral: {max_pct}%)",
45        registros_afectados=int(df[columna].isna().sum()),
46    )
47
48def check_sin_duplicados(df: pd.DataFrame, columnas: list[str]) -> ResultadoValidacion:
49    """Verifica que no haya duplicados por las columnas clave."""
50    n_duplicados = df.duplicated(subset=columnas).sum()
51    return ResultadoValidacion(
52        nombre=f"sin_duplicados_{'_'.join(columnas)}",
53        paso=n_duplicados == 0,
54        mensaje=f"Duplicados por {columnas}: {n_duplicados}",
55        registros_afectados=n_duplicados,
56    )
57
58def check_rango(df: pd.DataFrame, columna: str, min_val: float, max_val: float) -> ResultadoValidacion:
59    """Verifica que los valores estén dentro de un rango."""
60    fuera = ((df[columna] < min_val) | (df[columna] > max_val)).sum()
61    return ResultadoValidacion(
62        nombre=f"rango_{columna}",
63        paso=fuera == 0,
64        mensaje=f"{columna}: {fuera} valores fuera de [{min_val}, {max_val}]",
65        registros_afectados=int(fuera),
66    )

Funciones puras que retornan objetos estructurados — fáciles de testear y combinar

### Pattern 2: El validador configurable

En la práctica, querrás validar muchos datasets diferentes con reglas diferentes. En vez de escribir funciones sueltas cada vez, crea un validador configurable: le pasas un DataFrame y una lista de reglas, y él ejecuta todas las validaciones y te da un informe.

1class ValidadorDataFrame:
2    """Validador configurable para DataFrames."""
3    
4    def __init__(self, nombre_dataset: str):
5        self.nombre = nombre_dataset
6        self.checks: list[Callable] = []
7    
8    def agregar_check(self, check_fn: Callable):
9        """Agrega una función de validación."""
10        self.checks.append(check_fn)
11        return self  # Para encadenar
12    
13    def ejecutar(self, df: pd.DataFrame) -> InformeValidacion:
14        """Ejecuta todos los checks y retorna el informe."""
15        informe = InformeValidacion()
16        for check_fn in self.checks:
17            resultado = check_fn(df)
18            informe.resultados.append(resultado)
19        return informe
20
21# Uso: configurar validador para tabla de pedidos
22validador_pedidos = ValidadorDataFrame("pedidos_diarios")
23validador_pedidos.agregar_check(lambda df: check_no_nulos(df, 'pedido_id'))
24validador_pedidos.agregar_check(lambda df: check_no_nulos(df, 'importe', max_pct=1.0))
25validador_pedidos.agregar_check(lambda df: check_sin_duplicados(df, ['pedido_id']))
26validador_pedidos.agregar_check(lambda df: check_rango(df, 'importe', 0.01, 50000))
27
28# Ejecutar
29pedidos = pd.DataFrame({
30    'pedido_id': [1, 2, 3, 3, 4],
31    'importe': [50.0, None, 120.0, 120.0, -5.0],
32})
33
34informe = validador_pedidos.ejecutar(pedidos)
35print(f"Dataset: {validador_pedidos.nombre}")
36print(f"Resultado: {informe.resumen()}")
37for r in informe.resultados:
38    estado = '✅' if r.paso else '🔴'
39    print(f"  {estado} {r.nombre}: {r.mensaje}")

Un validador configurable te permite reutilizar checks y mantener las reglas separadas del código

### Pattern 3: Decoradores de validación

Los decoradores son una herramienta elegante de Python para "envolver" funciones con lógica adicional. Aplicados a validación, permiten decir: "antes de ejecutar esta función de transformación, valida la entrada. Después de ejecutarla, valida la salida." Es como poner un guardia en la puerta de entrada Y de salida de tu función.

1import functools
2import logging
3
4logging.basicConfig(level=logging.INFO)
5logger = logging.getLogger('pipeline')
6
7def validar_entrada(checks: list[Callable]):
8    """Decorador que valida el DataFrame de entrada antes de procesarlo."""
9    def decorator(func):
10        @functools.wraps(func)
11        def wrapper(df: pd.DataFrame, *args, **kwargs):
12            logger.info(f"Validando entrada para {func.__name__}...")
13            
14            for check_fn in checks:
15                resultado = check_fn(df)
16                if not resultado.paso and resultado.severidad == 'ERROR':
17                    raise ValueError(
18                        f"Validación fallida en {func.__name__}: "
19                        f"{resultado.nombre} — {resultado.mensaje}"
20                    )
21                elif not resultado.paso:
22                    logger.warning(f"⚠️ {resultado.nombre}: {resultado.mensaje}")
23            
24            logger.info(f"✅ Entrada validada. Ejecutando {func.__name__}...")
25            return func(df, *args, **kwargs)
26        return wrapper
27    return decorator
28
29def validar_salida(checks: list[Callable]):
30    """Decorador que valida el DataFrame de salida después de procesarlo."""
31    def decorator(func):
32        @functools.wraps(func)
33        def wrapper(*args, **kwargs):
34            resultado_df = func(*args, **kwargs)
35            
36            logger.info(f"Validando salida de {func.__name__}...")
37            for check_fn in checks:
38                resultado = check_fn(resultado_df)
39                if not resultado.paso and resultado.severidad == 'ERROR':
40                    raise ValueError(
41                        f"Salida inválida de {func.__name__}: "
42                        f"{resultado.nombre} — {resultado.mensaje}"
43                    )
44            
45            logger.info(f"✅ Salida validada para {func.__name__}")
46            return resultado_df
47        return wrapper
48    return decorator
49
50# Uso del decorador
51@validar_entrada([
52    lambda df: check_no_nulos(df, 'importe'),
53    lambda df: check_rango(df, 'importe', 0, 100000),
54])
55@validar_salida([
56    lambda df: check_no_nulos(df, 'importe_con_iva'),
57])
58def calcular_iva(df: pd.DataFrame) -> pd.DataFrame:
59    """Añade columna de IVA al DataFrame de ventas."""
60    df = df.copy()
61    df['importe_con_iva'] = df['importe'] * 1.21
62    return df

Decoradores: validación automática en entrada y salida sin ensuciar la lógica de negocio

Consejo de senior: los decoradores de validación son GENIALES para pipelines de producción. Separas la lógica de transformación (el "qué hace") de la validación (el "qué espero"). Cuando cambias una regla de validación, no tocas la función. Cuando cambias la transformación, no tocas las validaciones. Separación de responsabilidades en acción.

### Pattern 4: Validación con pytest — testing de datos

Ya conoces pytest de la Skill 9. Aquí lo usamos para algo que quizás no esperabas: testear DATOS, no código. La idea es escribir tests que se ejecutan contra los datos de producción (o un sample) y fallan si los datos no cumplen las expectativas. Es como TDD pero para datos.

1# tests/test_calidad_pedidos.py
2import pandas as pd
3import pytest
4
5@pytest.fixture
6def pedidos():
7    """Carga los pedidos del día (en producción, lee del warehouse)."""
8    return pd.read_csv('data/pedidos_hoy.csv')
9
10class TestCalidadPedidos:
11    """Suite de tests de calidad para la tabla de pedidos."""
12    
13    def test_no_vacio(self, pedidos):
14        """El pipeline debe producir al menos 100 pedidos diarios."""
15        assert len(pedidos) >= 100, f"Solo {len(pedidos)} pedidos — ¿falló la ingesta?"
16    
17    def test_sin_duplicados(self, pedidos):
18        """Cada pedido_id debe ser único."""
19        duplicados = pedidos['pedido_id'].duplicated().sum()
20        assert duplicados == 0, f"{duplicados} pedidos duplicados"
21    
22    def test_importes_positivos(self, pedidos):
23        """Todos los importes deben ser positivos."""
24        negativos = (pedidos['importe'] <= 0).sum()
25        assert negativos == 0, f"{negativos} importes no positivos"
26    
27    def test_completitud_email(self, pedidos):
28        """Máximo 2% de emails pueden ser NULL."""
29        pct_nulos = pedidos['email'].isna().mean()
30        assert pct_nulos <= 0.02, f"{pct_nulos:.1%} de emails son NULL (umbral: 2%)"
31    
32    def test_fechas_razonables(self, pedidos):
33        """Las fechas deben ser del día actual o el anterior."""
34        pedidos['fecha'] = pd.to_datetime(pedidos['fecha_pedido'])
35        hoy = pd.Timestamp.now().normalize()
36        ayer = hoy - pd.Timedelta(days=1)
37        fuera_rango = ((pedidos['fecha'] < ayer) | (pedidos['fecha'] > hoy)).sum()
38        assert fuera_rango == 0, f"{fuera_rango} pedidos con fecha fuera de rango"
39
40# Ejecutar: pytest tests/test_calidad_pedidos.py -v

pytest para datos: cada test verifica una dimensión de calidad. Si falla, el pipeline se detiene.

### Cuándo usar cada pattern

Funciones validadoras puras: cuando necesitas validación ad-hoc, scripts rápidos, o construir tu propio framework
Validador configurable: cuando tienes múltiples datasets con reglas diferentes y quieres gestionar las reglas como configuración
Decoradores: cuando quieres validación transparente en funciones de pipeline sin modificar su lógica interna
pytest: cuando quieres integrar la validación de datos en tu CI/CD pipeline, con reportes y failures claros

Lo que le diría a mi yo de hace 5 años: empieza SIMPLE. Un archivo validate.py con 3-4 funciones que verifican lo básico (nulos, duplicados, rangos) es infinitamente mejor que nada. Puedes sofisticarlo después. No dejes que la parálisis del análisis ("necesito un framework") te impida poner al menos un check básico HOY.

Con estos patterns tienes todas las herramientas para validar datos con Python puro. En la próxima lección vamos a ver Great Expectations, que toma estos mismos conceptos y los lleva a otro nivel: configuración declarativa, reportes HTML automáticos, integración con orquestadores y mucho más.

## ejercicios

[01]

Construir un validador configurable para transacciones

El equipo de fraude te pide un validador para transacciones bancarias. Debe ser configurable (las reglas cambian según el tipo de cuenta) y generar un informe estructurado. Implementa el ValidadorTransacciones.

💡 Resultado esperado

=== VALIDACIÓN DE TRANSACCIONES ===

✅ 1 OK | 🔴 3 errores | ⚠️ 1 warnings

Cargando editor...

[02]

Crear decoradores de validación para un pipeline ETL

Tienes 3 funciones de transformación en tu pipeline: limpiar_datos, enriquecer_datos y agregar_metricas. Escribe decoradores que validen la entrada y salida de cada función, y que lancen una excepción si algo falla.

💡 Resultado esperado

✅ Entrada validada para limpiar_datos (8 filas)

💥 Pipeline abortado: [limpiar_datos] Perdida excesiva: 37.5% (8 → 5). Máximo: 20.0%

Cargando editor...

[03]

Escribir tests de calidad con pytest

Escribe una suite de tests con pytest que valide la calidad del dataset de ventas diario. Los tests deben cubrir: volumen mínimo, completitud, rango de valores y frescura. Usa fixtures de pytest.

Cargando editor...

Regístrate para guardar tu progreso.

## comentarios

Reporta erratas, ayuda a otros o comparte tu opinión. Sé constructivo.

Inicia sesión para comentar y responder.

cargando comentarios...