SQL com Python: consultando bancos de dados com Pandas e SQLAlchemy

Conectar Python a bancos de dados SQL abre acesso direto a dados corporativos sem precisar exportar CSVs manualmente. Com SQLAlchemy e Pandas, você escreve SQL familiar e recebe um DataFrame pronto para análise — sem ORM obrigatório, sem camada extra de abstração quando não precisa.

Instalação

pip install sqlalchemy pandas psycopg2-binary  # PostgreSQL
# ou:
pip install sqlalchemy pandas pymysql          # MySQL
# SQLite já vem com Python (sem instalação adicional)

Conectando ao banco

from sqlalchemy import create_engine

# SQLite local (ótimo para testes):
engine = create_engine("sqlite:///dados.db")

# PostgreSQL:
engine = create_engine("postgresql+psycopg2://usuario:senha@localhost:5432/banco")

# MySQL:
engine = create_engine("mysql+pymysql://usuario:senha@localhost:3306/banco")

# Boas práticas: ler credenciais de variáveis de ambiente
import os
DB_URL = os.environ.get("DATABASE_URL")
engine = create_engine(DB_URL)

Lendo dados com pandas.read_sql()

import pandas as pd

# Query simples:
df = pd.read_sql("SELECT * FROM vendas", engine)

# Query com filtros:
df = pd.read_sql('''
    SELECT
        v.id_pedido,
        c.nome AS cliente,
        v.valor,
        v.data_pedido
    FROM vendas v
    JOIN clientes c ON v.id_cliente = c.id
    WHERE v.data_pedido >= '2026-01-01'
      AND v.valor > 500
    ORDER BY v.data_pedido DESC
''', engine)

# Com parâmetros (evita SQL injection):
regiao = "Sul"
df = pd.read_sql(
    "SELECT * FROM vendas WHERE regiao = %(regiao)s",
    engine,
    params={"regiao": regiao}
)

Escrevendo DataFrames no banco

# Criar/substituir tabela:
df.to_sql("vendas_processadas", engine, if_exists="replace", index=False)

# Apenas adicionar novos registros (append):
df_novos.to_sql("vendas_processadas", engine, if_exists="append", index=False)

# Chunk para grandes volumes:
df.to_sql("vendas", engine, if_exists="replace", index=False,
          chunksize=10000, method="multi")

Executando DDL e DML com connection

from sqlalchemy import text

with engine.connect() as conn:
    # Criar tabela:
    conn.execute(text('''
        CREATE TABLE IF NOT EXISTS log_processamento (
            id INTEGER PRIMARY KEY,
            data_exec TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            registros INTEGER,
            status TEXT
        )
    '''))
    # Inserir registro de log:
    conn.execute(text(
        "INSERT INTO log_processamento (registros, status) VALUES (:n, :s)"
    ), {"n": len(df), "s": "OK"})
    conn.commit()

Contexto de conexão para múltiplas operações

# Com context manager, a conexão é fechada automaticamente:
with engine.begin() as conn:
    df1 = pd.read_sql("SELECT * FROM pedidos WHERE status = 'Pendente'", conn)
    df2 = pd.read_sql("SELECT * FROM clientes", conn)

resultado = pd.merge(df1, df2, on="id_cliente")

Perguntas frequentes

engine.connect() vs engine.begin(): qual usar?

engine.connect() requer conn.commit() manual. engine.begin() faz commit automático ao sair do bloco (ou rollback se houver exceção). Para operações de escrita, prefira engine.begin().

Como evitar carregar uma tabela inteira em memória?

Use pd.read_sql(query, engine, chunksize=50000) — retorna um iterador de DataFrames, processado um bloco por vez. Ideal para tabelas com milhões de linhas.