Como criar um ETL simples com Python e PostgreSQL — do zero ao dado limpo

# extract/api_source.py import requests import pandas as pd from dotenv import load_dotenv import os load_dotenv() def extract_from_api(endpoint: str, params: dict = {}) -> pd.DataFrame: """Extrai dados de uma API REST e retorna DataFrame.""" headers = {"Authorization": f"Bearer {os.getenv('API_TOKEN')}"} response = requests.get(endpoint, headers=headers, params=params, timeout=30) response.raise_for_status() data = response.json() # Normaliza JSON aninhado automaticamente return pd.json_normalize(data if isinstance(data, list) else data.get("items", [])) # extract/csv_source.py def extract_from_csv(filepath: str) -> pd.DataFrame: """Extrai de CSV com detecção automática de encoding.""" for encoding in ["utf-8", "latin-1", "cp1252"]: try: return pd.read_csv(filepath, encoding=encoding) except UnicodeDecodeError: continue raise ValueError(f"Não foi possível ler {filepath} com os encodings tentados")

# transform/normalize.py import pandas as pd def normalize_customers(df: pd.DataFrame) -> pd.DataFrame: """Normaliza tabela de clientes — aplicar após extract.""" df = df.copy() # Padronizar nomes de colunas df.columns = [col.lower().strip().replace(" ", "_") for col in df.columns] # Limpar CPF/CNPJ (remover pontuação) if "documento" in df.columns: df["documento"] = df["documento"].str.replace(r"[.-/]", "", regex=True) # Normalizar datas (múltiplos formatos possíveis) if "data_cadastro" in df.columns: df["data_cadastro"] = pd.to_datetime(df["data_cadastro"], dayfirst=True, errors="coerce") # Remover duplicatas (pelo documento mais recente) if "documento" in df.columns: df = df.sort_values("data_cadastro", ascending=False) df = df.drop_duplicates(subset=["documento"], keep="first") # Preencher valores nulos df["cidade"] = df.get("cidade", pd.Series()).fillna("Não informado") # Adicionar timestamp de processamento df["etl_loaded_at"] = pd.Timestamp.utcnow() return df

# load/postgres.py from sqlalchemy import create_engine, text import pandas as pd import os def get_engine(): url = (f"postgresql+psycopg2://{os.getenv('PG_USER')}:{os.getenv('PG_PASSWORD')}" f"@{os.getenv('PG_HOST')}:{os.getenv('PG_PORT')}/{os.getenv('PG_DATABASE')}") return create_engine(url, pool_pre_ping=True) def upsert_dataframe(df: pd.DataFrame, table: str, pk_columns: list[str]) -> int: """ Upsert: INSERT ON CONFLICT DO UPDATE Idempotente — rodar 2x não gera duplicatas. Retorna o número de linhas afetadas. """ engine = get_engine() # Gravar em tabela temporária temp_table = f"_temp_{table}" df.to_sql(temp_table, engine, if_exists="replace", index=False) # Colunas para atualizar (tudo exceto as PKs) update_cols = [c for c in df.columns if c not in pk_columns] update_stmt = ", ".join(f"{c} = EXCLUDED.{c}" for c in update_cols) pk_stmt = ", ".join(pk_columns) upsert_sql = f""" INSERT INTO {table} SELECT * FROM {temp_table} ON CONFLICT ({pk_stmt}) DO UPDATE SET {update_stmt}; DROP TABLE {temp_table}; """ with engine.connect() as conn: result = conn.execute(text(upsert_sql)) conn.commit() return result.rowcount

Para rodar o ETL automaticamente todo dia às 6h da manhã:

# Editar crontab
crontab -e

# Adicionar linha (roda às 06:00 todo dia)
0 6 * * * cd /opt/etl-projeto && /usr/bin/python3 main.py >> /var/log/etl.log 2>&1

O >>> acumula logs sem sobrescrever. Monitore com tail -f /var/log/etl.log.

Para alertas de falha, adicione ao main.py:

try:
run_pipeline()
except Exception as e:
requests.post(SLACK_WEBHOOK, json={"text": f"❌ ETL falhou: {e}"})