Contratos de datos: definir qué esperas ANTES de que lleguen

1# Ejemplo de contrato de datos en Python (como diccionario/YAML)
2contrato_pedidos = {
3    "nombre": "pedidos_diarios",
4    "version": "2.1.0",
5    "productor": {
6        "equipo": "backend-ecommerce",
7        "contacto": "#canal-backend-datos",
8        "sla_actualizacion_horas": 1,
9    },
10    "consumidores": [
11        {"equipo": "data-engineering", "uso": "pipeline de métricas"},
12        {"equipo": "finance", "uso": "facturación"},
13    ],
14    "schema": {
15        "pedido_id": {"tipo": "string", "obligatorio": True, "unico": True, "formato": "PED-\\d{6}"},
16        "cliente_id": {"tipo": "integer", "obligatorio": True, "min": 1},
17        "importe_total": {"tipo": "float", "obligatorio": True, "min": 0.01, "max": 100000},
18        "estado": {"tipo": "string", "obligatorio": True, "valores_validos": ["pendiente", "pagado", "enviado", "entregado", "cancelado"]},
19        "fecha_pedido": {"tipo": "datetime", "obligatorio": True, "formato": "ISO8601"},
20        "ciudad": {"tipo": "string", "obligatorio": False, "max_nulos_pct": 5},
21        "descuento_pct": {"tipo": "float", "obligatorio": False, "min": 0, "max": 100},
22    },
23    "volumen": {
24        "min_registros_diarios": 500,
25        "max_registros_diarios": 50000,
26    },
27    "freshness": {
28        "frecuencia": "cada hora",
29        "max_retraso_minutos": 90,
30    },
31}

Un contrato de datos legible que ambas partes (productor y consumidor) pueden verificar automáticamente

1from dataclasses import dataclass, field
2import pandas as pd
3import re
4from typing import Any
5
6@dataclass
7class CampoContrato:
8    """Especificación de un campo en el contrato."""
9    nombre: str
10    tipo: str          # 'string', 'integer', 'float', 'datetime'
11    obligatorio: bool = True
12    unico: bool = False
13    min_valor: float | None = None
14    max_valor: float | None = None
15    valores_validos: list[str] | None = None
16    formato_regex: str | None = None
17    max_nulos_pct: float = 0.0  # % máximo de nulos permitido
18
19@dataclass
20class ContratoSchema:
21    """Contrato completo para un dataset."""
22    nombre: str
23    version: str
24    productor: str
25    campos: list[CampoContrato]
26    min_registros: int = 0
27    max_registros: int | None = None
28
29    def validar(self, df: pd.DataFrame) -> dict:
30        """Valida un DataFrame contra este contrato."""
31        violaciones = []
32        
33        # Verificar columnas esperadas
34        campos_esperados = {c.nombre for c in self.campos if c.obligatorio}
35        campos_presentes = set(df.columns)
36        faltantes = campos_esperados - campos_presentes
37        if faltantes:
38            violaciones.append(f"Columnas faltantes: {faltantes}")
39        
40        # Verificar volumen
41        if len(df) < self.min_registros:
42            violaciones.append(f"Volumen insuficiente: {len(df)} < {self.min_registros}")
43        if self.max_registros and len(df) > self.max_registros:
44            violaciones.append(f"Volumen excesivo: {len(df)} > {self.max_registros}")
45        
46        # Verificar cada campo
47        for campo in self.campos:
48            if campo.nombre not in df.columns:
49                continue
50            
51            col = df[campo.nombre]
52            
53            # Nulos
54            pct_nulos = col.isna().mean() * 100
55            if pct_nulos > campo.max_nulos_pct:
56                violaciones.append(
57                    f"{campo.nombre}: {pct_nulos:.1f}% nulos (máx: {campo.max_nulos_pct}%)"
58                )
59            
60            # Unicidad
61            if campo.unico and col.dropna().duplicated().any():
62                n_dup = col.dropna().duplicated().sum()
63                violaciones.append(f"{campo.nombre}: {n_dup} duplicados (debe ser único)")
64            
65            # Rango
66            if campo.min_valor is not None:
67                fuera = (col.dropna() < campo.min_valor).sum()
68                if fuera > 0:
69                    violaciones.append(f"{campo.nombre}: {fuera} valores < {campo.min_valor}")
70            if campo.max_valor is not None:
71                fuera = (col.dropna() > campo.max_valor).sum()
72                if fuera > 0:
73                    violaciones.append(f"{campo.nombre}: {fuera} valores > {campo.max_valor}")
74            
75            # Valores válidos
76            if campo.valores_validos:
77                invalidos = ~col.dropna().isin(campo.valores_validos)
78                if invalidos.any():
79                    violaciones.append(
80                        f"{campo.nombre}: {invalidos.sum()} valores fuera del conjunto permitido"
81                    )
82        
83        return {
84            'contrato': self.nombre,
85            'version': self.version,
86            'cumple': len(violaciones) == 0,
87            'violaciones': violaciones,
88            'total_violaciones': len(violaciones),
89        }
90
91# Definir contrato para pedidos
92contrato = ContratoSchema(
93    nombre="pedidos_diarios",
94    version="1.0.0",
95    productor="equipo-backend",
96    min_registros=10,
97    campos=[
98        CampoContrato("pedido_id", "string", obligatorio=True, unico=True),
99        CampoContrato("importe", "float", obligatorio=True, min_valor=0.01, max_valor=50000),
100        CampoContrato("estado", "string", obligatorio=True,
101                       valores_validos=["pendiente", "pagado", "enviado", "entregado"]),
102        CampoContrato("email", "string", obligatorio=True, max_nulos_pct=5.0),
103    ],
104)
105
106# Validar datos contra el contrato
107datos = pd.DataFrame({
108    'pedido_id': ['P001', 'P002', 'P003', 'P003', 'P005'],
109    'importe': [50.0, -10.0, 300.0, 150.0, 75000.0],
110    'estado': ['pagado', 'enviado', 'devuelto', 'pendiente', 'pagado'],
111    'email': ['a@m.com', None, 'c@m.com', None, 'e@m.com'],
112})
113
114resultado = contrato.validar(datos)
115print(f"Contrato: {resultado['contrato']} v{resultado['version']}")
116print(f"Cumple: {'✅' if resultado['cumple'] else '🔴 NO'}\n")
117for v in resultado['violaciones']:
118    print(f"  ❌ {v}")

Un sistema de contratos implementado en Python puro — verificable automáticamente

1# Ejemplo de evolución de un contrato
2historial_versiones = [
3    {
4        "version": "1.0.0",
5        "fecha": "2024-01-01",
6        "cambios": "Versión inicial del contrato de pedidos",
7    },
8    {
9        "version": "1.1.0",
10        "fecha": "2024-03-15",
11        "tipo": "MINOR",
12        "cambios": "Añadido campo opcional 'canal_venta' (web/app/tienda)",
13        "impacto": "Ninguno — campo nuevo y opcional",
14    },
15    {
16        "version": "2.0.0",
17        "fecha": "2024-06-01",
18        "tipo": "MAJOR ⚠️",
19        "cambios": "Campo 'total' renombrado a 'importe_total'. Campo 'status' cambia de español a códigos (1=pendiente, 2=pagado...)",
20        "impacto": "BREAKING — todos los consumidores deben actualizar",
21        "deadline_migracion": "2024-06-15",
22    },
23]
24
25for v in historial_versiones:
26    tipo_badge = f" [{v.get('tipo', 'INITIAL')}]" if 'tipo' in v else ""
27    print(f"v{v['version']}{tipo_badge} — {v['fecha']}")
28    print(f"  {v['cambios']}")
29    if 'impacto' in v:
30        print(f"  Impacto: {v['impacto']}")
31    print()

Un historial de cambios claro permite a los consumidores prepararse para breaking changes