👁️ Monitoramento e Gestão de Dados — Passo a Passo

O generated_at é o GPS temporal de cada tabela-resumo. Sem ele, você não sabe se está olhando para dados de ontem ou de seis meses atrás — e o agente também não.

from datetime import datetime, timezone

# Ao criar qualquer tabela-resumo:
generated_at = datetime.now(timezone.utc).isoformat()

conn.execute(f"""
  CREATE OR REPLACE TABLE revenue_by_client_month AS
  SELECT
    customer_email,
    DATE_TRUNC('month', payment_date) AS month,
    SUM(amount_usd) AS total_revenue_usd,
    COUNT(*) AS transaction_count,
    '{generated_at}'::TIMESTAMP AS generated_at  -- <-- OBRIGATÓRIO
  FROM stripe_payments
  WHERE status = 'succeeded'
  GROUP BY customer_email, DATE_TRUNC('month', payment_date)
""")

# Também salvar metadados separadamente
conn.execute(f"""
  INSERT INTO pipeline_runs (table_name, generated_at, row_count, status)
  VALUES (
    'revenue_by_client_month',
    '{generated_at}',
    (SELECT COUNT(*) FROM revenue_by_client_month),
    'SUCCESS'
  )
""")

Cada reprocessamento é uma nova versão. Sem versionamento, quando o cliente diz "os números mudaram desde ontem", você não tem como comparar a versão atual com a anterior.

CREATE TABLE IF NOT EXISTS pipeline_runs (
  run_id VARCHAR DEFAULT gen_random_uuid(),
  table_name VARCHAR NOT NULL,
  generated_at TIMESTAMP NOT NULL,
  row_count INTEGER,
  status VARCHAR,   -- 'SUCCESS', 'FAILED', 'PARTIAL'
  trigger_reason VARCHAR,  -- 'scheduled', 'manual', 'data_quality_fail'
  notes TEXT,
  PRIMARY KEY (run_id)
);

-- Comparar duas versões
SELECT
  r1.generated_at AS versao_atual,
  r2.generated_at AS versao_anterior,
  r1.row_count - r2.row_count AS delta_linhas
FROM pipeline_runs r1
JOIN pipeline_runs r2 ON r1.table_name = r2.table_name
WHERE r1.table_name = 'revenue_by_client_month'
ORDER BY r1.generated_at DESC
LIMIT 2

Detectar antes que o agente detecte — ou pior, antes que o cliente detecte. Os 4 sinais clássicos cobrem 95% das falhas reais de pipelines em produção.

def verificar_qualidade(conn, table_name):
    checks = {}

    # Volume
    row_count = conn.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone()[0]
    checks['volume'] = {'valor': row_count, 'status': 'OK'}

    # Nulos em colunas críticas
    null_rate = conn.execute(f"""
        SELECT AVG(CASE WHEN total_revenue_usd IS NULL THEN 1.0 ELSE 0.0 END)
        FROM {table_name}
    """).fetchone()[0]
    checks['null_rate'] = {
        'valor': f"{null_rate*100:.1f}%",
        'status': 'FAIL' if null_rate > 0.05 else 'OK'
    }

    # Range de valores
    neg_revenue = conn.execute(f"""
        SELECT COUNT(*) FROM {table_name}
        WHERE total_revenue_usd < 0
    """).fetchone()[0]
    checks['range'] = {
        'valor': f"{neg_revenue} negativos",
        'status': 'FAIL' if neg_revenue > 0 else 'OK'
    }

    return checks

SLA (Service Level Agreement) de dados é o contrato interno que transforma "acho que tá funcionando" em "está dentro do contrato" ou "está fora do contrato".

Monitoramento manual é insustentável. Um agente que detecta e reage automaticamente elimina o trabalho repetitivo e garante SLA mesmo durante fins de semana.

# verify_pipeline.py — saída JSON obrigatória
import json, sys
checks = verificar_qualidade(conn, 'revenue_by_client_month')
failed = [k for k, v in checks.items() if v['status'] == 'FAIL']
result = {'status': 'FAIL' if failed else 'OK', 'failed_checks': failed, 'details': checks}
print(json.dumps(result))
sys.exit(1 if failed else 0)

Documentação bem feita permite que outro engenheiro (ou o próprio agente) mantenha o pipeline sem precisar de você. É a diferença entre um projeto sustentável e um projeto que só você consegue operar.