Python para Data Engineering

Python Esencial para Data Engineering

Antes de las librerías, el dominio de Python puro es fundamental. Un Data Engineer escribe código de producción — limpio, testeable, mantenible y eficiente.

Patrones Pythónicos Clave

"""
Python patterns esenciales para Data Engineering
Python 3.11+ features incluidos
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Generator, Iterator, Any
from pathlib import Path
from contextlib import contextmanager
from functools import lru_cache, partial
import itertools
import json

# ── DATACLASSES: Estructuras de datos tipadas ──────────────────────────────
@dataclass
class DataPipelineConfig:
    """Configuración de pipeline con defaults y validación."""
    source_url: str
    batch_size: int = 1000
    max_retries: int = 3
    timeout_seconds: float = 30.0
    output_path: Path = field(default_factory=lambda: Path("./output"))
    tags: list[str] = field(default_factory=list)

    def __post_init__(self):
        # Validación al crear
        if self.batch_size <= 0:
            raise ValueError(f"batch_size debe ser positivo, got {self.batch_size}")
        if not self.source_url.startswith(("http://", "https://", "gs://", "s3://")):
            raise ValueError(f"URL de fuente inválida: {self.source_url}")
        self.output_path.mkdir(parents=True, exist_ok=True)

# ── GENERATORS: Procesamiento de datos en streaming sin cargar en memoria ──
def read_csv_in_chunks(filepath: Path, chunk_size: int = 1000) -> Generator[list[dict], None, None]:
    """Genera chunks de CSV sin cargar el archivo completo en memoria."""
    with open(filepath, 'r', encoding='utf-8') as f:
        import csv
        reader = csv.DictReader(f)
        batch = []
        for i, row in enumerate(reader):
            batch.append(row)
            if len(batch) >= chunk_size:
                yield batch
                batch = []
        if batch:  # Último batch (puede ser menor que chunk_size)
            yield batch

# Uso: procesa millones de filas con memoria constante
for chunk in read_csv_in_chunks(Path("huge_file.csv"), chunk_size=5000):
    process_batch(chunk)

# ── CONTEXT MANAGERS: Gestión automática de recursos ──────────────────────
@contextmanager
def timer(operation_name: str) -> Iterator[None]:
    """Context manager para medir tiempo de operaciones."""
    import time
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        print(f"⏱️ {operation_name}: {elapsed:.3f}s")

with timer("Data extraction"):
    data = extract_from_api()

# ── TYPE HINTS AVANZADOS (Python 3.10+) ────────────────────────────────────
from typing import TypeAlias, TypeVar

DataRow: TypeAlias = dict[str, Any]
T = TypeVar('T')

def transform_rows(
    rows: list[DataRow],
    *transformers: Callable[[DataRow], DataRow],
    filter_fn: Callable[[DataRow], bool] | None = None
) -> list[DataRow]:
    """Aplica una cadena de transformaciones a filas de datos."""
    result = rows
    for transform in transformers:
        result = [transform(row) for row in result]
    if filter_fn:
        result = [row for row in result if filter_fn(row)]
    return result

# ── MATCH STATEMENT (Python 3.10+) ─────────────────────────────────────────
def handle_api_response(response_data: dict) -> str:
    """Maneja diferentes tipos de respuesta de API."""
    match response_data:
        case {"status": "success", "data": list() as items}:
            return f"✅ {len(items)} registros recibidos"
        case {"status": "error", "code": int(code), "message": str(msg)}:
            return f"❌ Error {code}: {msg}"
        case {"status": "partial", "data": items, "total": total}:
            return f"⚠️ Parcial: {len(items)}/{total} registros"
        case _:
            return f"❓ Respuesta desconocida: {response_data}"

Manejo de Errores y Retry Logic

"""
Patrones de resiliencia para pipelines de producción
"""
import time
import logging
from functools import wraps
from typing import TypeVar, Callable
from exceptions import PipelineError, TransientError

F = TypeVar('F', bound=Callable)
logger = logging.getLogger(__name__)

def retry_with_backoff(
    max_attempts: int = 3,
    initial_delay: float = 1.0,
    backoff_factor: float = 2.0,
    exceptions: tuple[type[Exception], ...] = (TransientError,)
):
    """Decorador que reintenta automáticamente ante errores transientes."""
    def decorator(func: F) -> F:
        @wraps(func)
        def wrapper(*args, **kwargs):
            delay = initial_delay
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_attempts - 1:
                        logger.error(f"Max retries ({max_attempts}) exceeded: {e}")
                        raise
                    logger.warning(
                        f"Attempt {attempt + 1}/{max_attempts} failed: {e}. "
                        f"Retrying in {delay:.1f}s..."
                    )
                    time.sleep(delay)
                    delay *= backoff_factor
        return wrapper
    return decorator

@retry_with_backoff(max_attempts=5, initial_delay=2.0)
def fetch_data_from_api(endpoint: str) -> dict:
    """Fetch con retry automático."""
    import requests
    response = requests.get(endpoint, timeout=30)
    response.raise_for_status()
    return response.json()

Pandas: El Estándar de la Industria

Pandas sigue siendo la librería más usada para manipulación de datos en Python. Dominar sus patrones avanzados es imprescindible.

"""
Pandas avanzado para Data Engineering
pandas 2.x con mejoras de performance
"""
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

# ── LECTURA EFICIENTE ───────────────────────────────────────────────────────
# Especificar dtypes al leer SIEMPRE (evita inferencia lenta)
dtypes = {
    'order_id':     'int64',
    'customer_id':  'int32',
    'total_amount': 'float32',  # float32 en lugar de float64 ahorra 50% memoria
    'status':       'category',  # category para strings repetitivos
}
df = pd.read_csv('orders.csv',
    dtype=dtypes,
    parse_dates=['order_date'],
    usecols=['order_id', 'customer_id', 'total_amount', 'status', 'order_date'],
    chunksize=None  # O leer en chunks si el archivo es enorme
)

# Leer con chunks para archivos grandes
reader = pd.read_csv('huge_file.csv', chunksize=50_000)
processed_chunks = []
for chunk in reader:
    clean_chunk = chunk.dropna(subset=['customer_id']).copy()
    processed_chunks.append(clean_chunk)
df = pd.concat(processed_chunks, ignore_index=True)

# ── TRANSFORMACIONES AVANZADAS ─────────────────────────────────────────────
# .assign() para transformaciones encadenadas (más limpio que múltiples =)
df_transformed = (df
    .assign(
        # Transformaciones de fecha
        year       = lambda x: x['order_date'].dt.year,
        month      = lambda x: x['order_date'].dt.month,
        day_of_week= lambda x: x['order_date'].dt.day_name(),
        week_num   = lambda x: x['order_date'].dt.isocalendar().week,
        # Campos calculados
        is_large_order = lambda x: x['total_amount'] > 500,
        discount_amount= lambda x: np.where(x['status'] == 'vip', 
                                             x['total_amount'] * 0.1, 0),
        # Normalización
        amount_normalized = lambda x: (x['total_amount'] - x['total_amount'].mean()) 
                                      / x['total_amount'].std()
    )
    .query("status != 'cancelled'")
    .drop_duplicates(subset=['order_id'])
)

# ── MERGE (JOIN) en Pandas ─────────────────────────────────────────────────
orders = pd.read_csv('orders.csv')
customers = pd.read_csv('customers.csv')
products = pd.read_csv('products.csv')

# JOIN múltiple encadenado
result = (orders
    .merge(customers, on='customer_id', how='left', suffixes=('', '_cust'))
    .merge(products, on='product_id', how='inner', suffixes=('', '_prod'))
)

# ── GROUPBY AVANZADO ────────────────────────────────────────────────────────
# agg() con múltiples funciones por columna
customer_stats = df.groupby('customer_id').agg(
    total_orders   = ('order_id', 'count'),
    total_revenue  = ('total_amount', 'sum'),
    avg_order      = ('total_amount', 'mean'),
    max_order      = ('total_amount', 'max'),
    first_purchase = ('order_date', 'min'),
    last_purchase  = ('order_date', 'max'),
    unique_statuses= ('status', 'nunique'),
).reset_index()

# Agregar métricas calculadas
customer_stats['lifespan_days'] = (
    customer_stats['last_purchase'] - customer_stats['first_purchase']
).dt.days

customer_stats['avg_days_between_orders'] = (
    customer_stats['lifespan_days'] / customer_stats['total_orders']
).round(1)

# ── WINDOW FUNCTIONS equivalentes en Pandas ────────────────────────────────
# transform: groupby pero mantiene el índice original (como WINDOW en SQL)
df['customer_rank'] = df.groupby('customer_id')['total_amount'].rank(
    method='dense', ascending=False
)
df['customer_total_rev'] = df.groupby('customer_id')['total_amount'].transform('sum')
df['pct_of_customer_rev'] = df['total_amount'] / df['customer_total_rev']

# Cumulative sum por grupo
df_sorted = df.sort_values(['customer_id', 'order_date'])
df_sorted['cumulative_spent'] = df_sorted.groupby('customer_id')['total_amount'].cumsum()

# ── OPTIMIZACIÓN DE MEMORIA ─────────────────────────────────────────────────
def optimize_dataframe(df: pd.DataFrame) -> pd.DataFrame:
    """Reduce el uso de memoria del DataFrame automáticamente."""
    original_mem = df.memory_usage(deep=True).sum() / 1024**2
    
    for col in df.columns:
        col_type = df[col].dtype
        if col_type == 'int64':
            # Reducir int64 → int32 o int16 si es posible
            if df[col].between(-32768, 32767).all():
                df[col] = df[col].astype('int16')
            elif df[col].between(-2147483648, 2147483647).all():
                df[col] = df[col].astype('int32')
        elif col_type == 'float64':
            df[col] = df[col].astype('float32')
        elif col_type == 'object':
            # Si hay poca cardinalidad, usar category
            if df[col].nunique() / len(df) < 0.5:
                df[col] = df[col].astype('category')
    
    new_mem = df.memory_usage(deep=True).sum() / 1024**2
    print(f"Memoria: {original_mem:.1f}MB → {new_mem:.1f}MB ({(1-new_mem/original_mem)*100:.0f}% reducción)")
    return df

Polars: El Futuro del Procesamiento en Python

Polars es la alternativa moderna a Pandas. Escrito en Rust, usa paralelismo nativo y operaciones lazy para ser 5-100x más rápido que Pandas en la mayoría de operaciones.

"""
Polars para Data Engineering
polars 0.20+
"""
import polars as pl
from polars import col, lit, when
from datetime import date

# ── LECTURA: Lazy por defecto ──────────────────────────────────────────────
# LazyFrame: define el plan sin ejecutar (como un query plan de SQL)
lf = pl.scan_csv("orders.csv",
    dtypes={
        "order_id": pl.Int64,
        "amount": pl.Float32,
        "status": pl.Categorical,
    },
    try_parse_dates=True
)

# La query se optimiza ANTES de ejecutar
result = (lf
    .filter(col("status") != "cancelled")
    .with_columns([
        col("order_date").dt.year().alias("year"),
        col("order_date").dt.month().alias("month"),
        (col("amount") * 1.21).alias("amount_with_tax"),
        when(col("amount") > 500).then(lit("large"))
        .when(col("amount") > 100).then(lit("medium"))
        .otherwise(lit("small")).alias("order_size"),
    ])
    .group_by(["year", "month", "status"])
    .agg([
        pl.count("order_id").alias("order_count"),
        pl.sum("amount").alias("total_revenue"),
        pl.mean("amount").alias("avg_amount"),
        pl.quantile("amount", 0.9).alias("p90_amount"),
    ])
    .sort(["year", "month"])
    .collect()  # Ejecutar el plan optimizado
)

# ── EXPRESIONES: más expresivas que Pandas ─────────────────────────────────
df = pl.read_csv("sales.csv")

# Múltiples transformaciones de una vez (más rápido que Pandas)
df_clean = df.with_columns([
    # String operations
    col("name").str.strip_chars().str.to_lowercase().alias("name_clean"),
    col("email").str.to_lowercase().alias("email_lower"),
    # Date operations
    col("order_date").str.strptime(pl.Date, "%Y-%m-%d"),
    # Conditional
    when(col("amount") >= 0).then(col("amount")).otherwise(lit(0)).alias("amount_positive"),
    # Arithmetic
    (col("price") * col("quantity") * (1 - col("discount_pct") / 100)).alias("net_amount"),
])

# ── GROUP BY: Mucho más rápido que Pandas ──────────────────────────────────
stats = df.group_by(["category", "year"]).agg([
    pl.count().alias("transactions"),
    pl.sum("amount").alias("revenue"),
    pl.n_unique("customer_id").alias("unique_customers"),
    pl.col("amount").mean().alias("avg_amount"),
    pl.col("amount").std().alias("std_amount"),
    pl.col("amount").quantile(0.5).alias("median"),
    pl.col("amount").filter(col("status") == "returned").sum().alias("returned_amount"),
])

# ── JOINS en Polars ────────────────────────────────────────────────────────
orders = pl.read_csv("orders.csv")
customers = pl.read_csv("customers.csv")

joined = orders.join(customers, on="customer_id", how="left")

# ── PANDAS vs POLARS: Comparativa de Performance ────────────────────────────
import time

# Pandas (1M rows)
import pandas as pd
df_pd = pd.DataFrame({"a": range(1_000_000), "b": [i % 100 for i in range(1_000_000)]})
t1 = time.perf_counter()
result_pd = df_pd.groupby("b")["a"].sum()
pandas_time = time.perf_counter() - t1

# Polars (mismo dataset)
df_pl = pl.DataFrame({"a": range(1_000_000), "b": [i % 100 for i in range(1_000_000)]})
t1 = time.perf_counter()
result_pl = df_pl.group_by("b").agg(pl.sum("a"))
polars_time = time.perf_counter() - t1

print(f"Pandas: {pandas_time:.3f}s")
print(f"Polars: {polars_time:.3f}s")
print(f"Speedup: {pandas_time/polars_time:.1f}x")  # Típicamente 10-50x

SQLAlchemy: Python ↔ Bases de Datos

"""
SQLAlchemy 2.0+ para Data Engineering
ORM + Core + Connection pooling
"""
from sqlalchemy import create_engine, text, Column, Integer, String, Float, DateTime
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
from sqlalchemy.dialects.postgresql import insert as pg_insert
from contextlib import contextmanager
import pandas as pd

# ── ENGINE: Connection pooling automático ─────────────────────────────────
engine = create_engine(
    "postgresql+psycopg2://user:password@host:5432/dbname",
    pool_size=10,           # Conexiones en el pool
    max_overflow=20,         # Conexiones adicionales máximas
    pool_pre_ping=True,      # Verifica conexiones antes de usarlas
    pool_recycle=3600,       # Recicla conexiones cada hora
    echo=False,              # True para debug (loguea SQL)
    connect_args={
        "connect_timeout": 10,
        "application_name": "my_data_pipeline",
    }
)

# ── ORM: Definir modelos ───────────────────────────────────────────────────
class Base(DeclarativeBase):
    pass

class Order(Base):
    __tablename__ = "orders"
    
    order_id    = Column(Integer, primary_key=True)
    customer_id = Column(Integer, nullable=False, index=True)
    amount      = Column(Float, nullable=False)
    status      = Column(String(20), default="pending", index=True)
    created_at  = Column(DateTime, nullable=False)

# Crear tablas
Base.metadata.create_all(engine)

# ── SESSION: Transacciones ─────────────────────────────────────────────────
SessionLocal = sessionmaker(bind=engine)

@contextmanager
def get_session():
    """Context manager para sesiones con rollback automático."""
    session = SessionLocal()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# Uso
with get_session() as session:
    new_order = Order(customer_id=1, amount=299.99, status="pending")
    session.add(new_order)
    # Commit automático al salir del context manager

# ── CORE: SQL expresivo ────────────────────────────────────────────────────
from sqlalchemy import select, func, and_, or_, update, delete

with engine.connect() as conn:
    # SELECT con filtros
    stmt = select(Order).where(
        and_(Order.status == "pending", Order.amount > 100)
    ).order_by(Order.created_at.desc()).limit(100)
    
    results = conn.execute(stmt).fetchall()
    
    # SQL raw con parámetros (seguro contra SQL injection)
    result = conn.execute(
        text("SELECT customer_id, SUM(amount) AS total FROM orders WHERE status = :status GROUP BY customer_id"),
        {"status": "delivered"}
    )

# ── PANDAS + SQLAlchemy: el dúo perfecto ──────────────────────────────────
# Leer query directamente en DataFrame
df = pd.read_sql_query(
    sql="SELECT * FROM orders WHERE order_date >= :since",
    con=engine,
    params={"since": "2026-01-01"}
)

# Escribir DataFrame a la base de datos
df.to_sql(
    name="fact_orders",
    con=engine,
    schema="analytics",
    if_exists="append",
    index=False,
    method="multi",      # Batch inserts
    chunksize=1000
)

# ── UPSERT: INSERT ON CONFLICT (PostgreSQL) ────────────────────────────────
def upsert_dataframe(df: pd.DataFrame, table_name: str, primary_keys: list[str]) -> int:
    """Upsert (INSERT or UPDATE) de un DataFrame en PostgreSQL."""
    rows = df.to_dict(orient='records')
    if not rows:
        return 0
    
    with engine.begin() as conn:
        stmt = pg_insert(table_name).values(rows)
        update_columns = {
            col: stmt.excluded[col]
            for col in df.columns
            if col not in primary_keys
        }
        stmt = stmt.on_conflict_do_update(
            index_elements=primary_keys,
            set_=update_columns
        )
        result = conn.execute(stmt)
    
    return result.rowcount

Ingesta desde APIs REST

"""
Patrones para consumir APIs REST en Data Engineering
httpx (async), requests, paginación, rate limiting
"""
import httpx
import asyncio
import time
from typing import AsyncGenerator
from dataclasses import dataclass

@dataclass
class APIConfig:
    base_url: str
    api_key: str
    rate_limit_rps: float = 10.0  # Requests por segundo
    timeout: float = 30.0

class DataAPIClient:
    def __init__(self, config: APIConfig):
        self.config = config
        self._last_request_time = 0.0
    
    def _rate_limit(self):
        """Throttle para respetar rate limits."""
        min_interval = 1.0 / self.config.rate_limit_rps
        elapsed = time.monotonic() - self._last_request_time
        if elapsed < min_interval:
            time.sleep(min_interval - elapsed)
        self._last_request_time = time.monotonic()
    
    @retry_with_backoff(max_attempts=3, exceptions=(httpx.HTTPError,))
    def get(self, endpoint: str, params: dict = None) -> dict:
        self._rate_limit()
        with httpx.Client(
            base_url=self.config.base_url,
            headers={"Authorization": f"Bearer {self.config.api_key}"},
            timeout=self.config.timeout
        ) as client:
            response = client.get(endpoint, params=params)
            response.raise_for_status()
            return response.json()
    
    def paginate(self, endpoint: str, page_size: int = 100) -> list[dict]:
        """Maneja paginación automáticamente."""
        all_data = []
        page = 1
        
        while True:
            response = self.get(endpoint, params={"page": page, "per_page": page_size})
            
            # Adaptar según el formato de la API
            if isinstance(response, list):
                data = response
                has_more = len(data) == page_size
            elif isinstance(response, dict):
                data = response.get("data", response.get("results", []))
                has_more = response.get("has_more", response.get("next") is not None)
            
            all_data.extend(data)
            
            if not has_more or not data:
                break
            
            page += 1
            print(f"  Página {page}: {len(all_data)} registros total")
        
        return all_data

# ── ASYNC: Múltiples APIs en paralelo ─────────────────────────────────────
async def fetch_all_endpoints(
    base_url: str,
    endpoints: list[str],
    api_key: str
) -> dict[str, list[dict]]:
    """Fetcha múltiples endpoints en paralelo (mucho más rápido que secuencial)."""
    async with httpx.AsyncClient(
        base_url=base_url,
        headers={"Authorization": f"Bearer {api_key}"},
        timeout=30.0
    ) as client:
        async def fetch_one(endpoint: str) -> tuple[str, list]:
            response = await client.get(endpoint)
            response.raise_for_status()
            return endpoint, response.json()
        
        results = await asyncio.gather(
            *[fetch_one(ep) for ep in endpoints],
            return_exceptions=True
        )
    
    return {
        endpoint: data
        for endpoint, data in results
        if not isinstance(data, Exception)
    }

# Uso: fetcha 10 endpoints en paralelo
endpoints = ["/users", "/orders", "/products", "/categories", "/reviews"]
all_data = asyncio.run(fetch_all_endpoints(BASE_URL, endpoints, API_KEY))
# En lugar de 10 * 2s = 20s, tarda ~2s (el tiempo del endpoint más lento)

Logging Profesional para Pipelines

"""
Logging estructurado para pipelines de producción
JSON logging para parseo automatizado por ELK/Grafana
"""
import logging
import json
import sys
from datetime import datetime, timezone

class StructuredFormatter(logging.Formatter):
    """Formatea logs como JSON para ingesta por sistemas de observabilidad."""
    
    def format(self, record: logging.LogRecord) -> str:
        log_data = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level":     record.levelname,
            "logger":    record.name,
            "message":   record.getMessage(),
            "module":    record.module,
            "function":  record.funcName,
            "line":      record.lineno,
        }
        # Agregar contexto adicional si existe
        if hasattr(record, 'pipeline_id'):
            log_data['pipeline_id'] = record.pipeline_id
        if hasattr(record, 'batch_id'):
            log_data['batch_id'] = record.batch_id
        if record.exc_info:
            log_data['exception'] = self.formatException(record.exc_info)
        
        return json.dumps(log_data)

def setup_logger(name: str, level: str = "INFO") -> logging.Logger:
    """Configura logger estructurado."""
    logger = logging.getLogger(name)
    logger.setLevel(getattr(logging, level.upper()))
    
    # Console handler (JSON en producción, legible en dev)
    handler = logging.StreamHandler(sys.stdout)
    if os.getenv("ENV", "dev") == "production":
        handler.setFormatter(StructuredFormatter())
    else:
        handler.setFormatter(logging.Formatter(
            "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s"
        ))
    
    logger.addHandler(handler)
    return logger

# Uso en un pipeline
logger = setup_logger("pipeline.orders")

class OrdersPipeline:
    def __init__(self):
        self.pipeline_id = f"orders_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
    
    def process_batch(self, batch_id: str, records: list) -> dict:
        extra = {"pipeline_id": self.pipeline_id, "batch_id": batch_id}
        logger.info(f"Processing batch: {len(records)} records", extra=extra)
        
        try:
            processed = [self.transform(r) for r in records]
            failed = [r for r in records if not self.validate(r)]
            
            logger.info(
                f"Batch complete: {len(processed)} processed, {len(failed)} failed",
                extra={**extra, "success_count": len(processed), "failed_count": len(failed)}
            )
            return {"processed": len(processed), "failed": len(failed)}
        
        except Exception as e:
            logger.error(f"Batch failed: {e}", exc_info=True, extra=extra)
            raise

Testing de Pipelines de Datos

"""
Testing con pytest para Data Engineering
pytest + pytest-mock + Great Expectations
"""
import pytest
import pandas as pd
import numpy as np
from unittest.mock import Mock, patch, MagicMock
from pathlib import Path

# ── UNIT TESTS: Funciones de transformación ────────────────────────────────
def normalize_email(email: str) -> str:
    return email.strip().lower() if email else None

def calculate_customer_ltv(orders_df: pd.DataFrame) -> pd.DataFrame:
    return orders_df.groupby('customer_id').agg(
        ltv=('amount', 'sum'),
        orders=('order_id', 'count')
    ).reset_index()

class TestNormalizeEmail:
    def test_lowercase_conversion(self):
        assert normalize_email("ANA@EXAMPLE.COM") == "ana@example.com"
    
    def test_strip_whitespace(self):
        assert normalize_email("  ana@example.com  ") == "ana@example.com"
    
    def test_none_input(self):
        assert normalize_email(None) is None
    
    def test_empty_string(self):
        assert normalize_email("") is None

class TestCalculateCustomerLTV:
    @pytest.fixture
    def sample_orders(self):
        return pd.DataFrame({
            'order_id':    [1, 2, 3, 4, 5],
            'customer_id': [101, 101, 102, 102, 103],
            'amount':      [100.0, 200.0, 50.0, 150.0, 300.0]
        })
    
    def test_correct_ltv_calculation(self, sample_orders):
        result = calculate_customer_ltv(sample_orders)
        customer_101 = result[result['customer_id'] == 101]
        assert customer_101['ltv'].values[0] == pytest.approx(300.0)
        assert customer_101['orders'].values[0] == 2
    
    def test_all_customers_present(self, sample_orders):
        result = calculate_customer_ltv(sample_orders)
        assert len(result) == 3  # 3 customers únicos
    
    def test_empty_dataframe(self):
        empty_df = pd.DataFrame(columns=['order_id', 'customer_id', 'amount'])
        result = calculate_customer_ltv(empty_df)
        assert len(result) == 0

# ── INTEGRATION TESTS: Mock de external services ───────────────────────────
class TestAPIIngestion:
    @patch('httpx.Client')
    def test_successful_api_fetch(self, mock_client_class):
        # Arrange
        mock_response = Mock()
        mock_response.json.return_value = [{"id": 1, "name": "Test"}]
        mock_response.status_code = 200
        mock_client = MagicMock()
        mock_client.__enter__.return_value.get.return_value = mock_response
        mock_client_class.return_value = mock_client
        
        # Act
        client = DataAPIClient(APIConfig(base_url="http://api.test", api_key="key"))
        result = client.get("/data")
        
        # Assert
        assert len(result) == 1
        assert result[0]["name"] == "Test"
    
    @patch('httpx.Client')
    def test_retry_on_server_error(self, mock_client_class):
        # Simulate 2 failures then success
        import httpx
        mock_get = Mock(side_effect=[
            httpx.HTTPStatusError("500", request=Mock(), response=Mock(status_code=500)),
            httpx.HTTPStatusError("503", request=Mock(), response=Mock(status_code=503)),
            Mock(json=lambda: {"data": "ok"}, status_code=200)
        ])
        mock_client = MagicMock()
        mock_client.__enter__.return_value.get = mock_get
        mock_client_class.return_value = mock_client
        
        # Should succeed on 3rd attempt
        result = client.get("/data")
        assert mock_get.call_count == 3

# ── DATA QUALITY TESTS: Validar schemas y constraints ──────────────────────
class TestDataQuality:
    def test_no_null_customer_ids(self, sample_orders):
        assert sample_orders['customer_id'].notna().all(), \
               "customer_id no debe tener valores nulos"
    
    def test_positive_amounts(self, sample_orders):
        assert (sample_orders['amount'] >= 0).all(), \
               "Los montos deben ser no-negativos"
    
    def test_unique_order_ids(self, sample_orders):
        assert sample_orders['order_id'].is_unique, \
               "order_id debe ser único"
    
    def test_schema_match(self, sample_orders):
        expected_dtypes = {
            'order_id': 'int64',
            'customer_id': 'int64',
            'amount': 'float64'
        }
        for col, expected_type in expected_dtypes.items():
            assert str(sample_orders[col].dtype) == expected_type, \
                   f"Columna {col}: esperado {expected_type}, got {sample_orders[col].dtype}"

Automatización de Pipelines

"""
Pipeline modular y automatizado con patrones de producción
"""
from abc import ABC, abstractmethod
from typing import Protocol, runtime_checkable

# ── PATTERN: Pipeline con etapas intercambiables ───────────────────────────
@runtime_checkable
class PipelineStage(Protocol):
    name: str
    def process(self, data: pd.DataFrame) -> pd.DataFrame: ...

class ExtractStage:
    name = "extract"
    def __init__(self, source: str):
        self.source = source
    def process(self, data: pd.DataFrame = None) -> pd.DataFrame:
        return pd.read_csv(self.source)

class CleanStage:
    name = "clean"
    def process(self, data: pd.DataFrame) -> pd.DataFrame:
        return (data
            .dropna(subset=['customer_id', 'amount'])
            .drop_duplicates()
            .assign(amount=lambda x: x['amount'].clip(lower=0))
        )

class Pipeline:
    def __init__(self, stages: list[PipelineStage]):
        self.stages = stages
        self.logger = setup_logger(f"pipeline.{id(self)}")
    
    def run(self) -> pd.DataFrame:
        data = None
        for stage in self.stages:
            self.logger.info(f"Running stage: {stage.name}")
            with timer(stage.name):
                data = stage.process(data)
            self.logger.info(f"Stage {stage.name} complete: {len(data)} rows")
        return data

# Uso: composición flexible de stages
pipeline = Pipeline([
    ExtractStage("s3://bucket/orders.csv"),
    CleanStage(),
    TransformStage(),
    LoadStage(engine=engine, table="fact_orders")
])
result = pipeline.run()