SQLAlchemy Core: consultas sem ORM

1. Introdução ao SQLAlchemy Core

SQLAlchemy é um dos frameworks de banco de dados mais poderosos do ecossistema Python, oferecendo duas abordagens distintas para interagir com bancos de dados relacionais: o ORM (Object-Relational Mapping) e o Core. Enquanto o ORM mapeia tabelas para classes Python e linhas para objetos, o Core trabalha diretamente com construções SQL em nível de schema e expressão.

O Core é ideal quando você precisa de:
- Controle fino sobre as consultas SQL geradas
- Performance máxima sem overhead de objetos
- Operações bulk e batch processing
- Integração com sistemas legados ou esquemas complexos

Para instalar o SQLAlchemy:

pip install sqlalchemy

2. Configuração do Engine e Conexão

O Engine é o ponto de partida para qualquer operação com SQLAlchemy Core. Ele gerencia o pool de conexões e dialoga com o banco de dados específico.

from sqlalchemy import create_engine

# Engine para SQLite (arquivo local)
engine = create_engine('sqlite:///meu_banco.db', echo=True)

# Engine para PostgreSQL
# engine = create_engine('postgresql://usuario:senha@localhost:5432/meudb')

# Usando context manager para gerenciar conexões
with engine.connect() as conn:
    resultado = conn.execute(text("SELECT 1"))
    print(resultado.fetchone())

O parâmetro echo=True ativa o logging das queries SQL geradas, essencial para depuração. O pooling padrão do SQLAlchemy gerencia automaticamente o reaproveitamento de conexões.

3. Definição de Esquemas com Metadata e Table

No Core, definimos tabelas como objetos Table associados a um objeto MetaData. Isso nos dá controle total sobre tipos, constraints e relacionamentos.

from sqlalchemy import MetaData, Table, Column, Integer, String, Float, DateTime
from datetime import datetime

metadata = MetaData()

# Definindo tabela de produtos
produtos = Table(
    'produtos',
    metadata,
    Column('id', Integer, primary_key=True, autoincrement=True),
    Column('nome', String(100), nullable=False),
    Column('preco', Float, nullable=False),
    Column('categoria', String(50)),
    Column('criado_em', DateTime, default=datetime.utcnow)
)

# Criando as tabelas no banco
metadata.create_all(engine)

Para refletir tabelas existentes do banco:

from sqlalchemy import inspect

inspector = inspect(engine)
tabelas = inspector.get_table_names()
print(f"Tabelas existentes: {tabelas}")

# Refletir uma tabela específica
from sqlalchemy import Table
usuarios = Table('usuarios', metadata, autoload_with=engine)

4. Construção de Consultas com select()

O Core constrói consultas usando objetos select() que podem ser combinados e modificados programaticamente.

from sqlalchemy import select, and_, or_

# Consulta básica
query = select(produtos)
resultado = conn.execute(query)

# Filtragem com where
query = select(produtos).where(produtos.c.preco > 100.0)

# Combinação de condições
query = select(produtos).where(
    and_(
        produtos.c.preco.between(50, 200),
        or_(
            produtos.c.categoria == 'Eletrônicos',
            produtos.c.categoria == 'Informática'
        )
    )
)

# Ordenação e paginação
query = select(produtos).order_by(
    produtos.c.preco.desc()
).limit(10).offset(20)

5. Execução de Consultas e Manipulação de Resultados

A execução de consultas retorna objetos Result que oferecem múltiplos métodos para acesso aos dados.

with engine.connect() as conn:
    # Executando consulta
    query = select(produtos)
    result = conn.execute(query)

    # fetchone() - pega um registro
    primeiro = result.fetchone()
    print(f"ID: {primeiro.id}, Nome: {primeiro.nome}")

    # fetchall() - pega todos
    result = conn.execute(query)
    todos = result.fetchall()

    # Mapeando para dicionários
    result = conn.execute(query)
    for row in result.mappings():
        print(dict(row))

    # scalar() - pega primeiro valor da primeira coluna
    query = select(produtos.c.preco).where(produtos.c.id == 1)
    preco = conn.execute(query).scalar()
    print(f"Preço do produto 1: {preco}")

6. Inserção, Atualização e Exclusão com Core

Operações de modificação usam objetos insert(), update() e delete().

from sqlalchemy import insert, update, delete

with engine.connect() as conn:
    # Inserção
    stmt = insert(produtos).values(
        nome='Notebook Dell',
        preco=4500.00,
        categoria='Informática'
    )
    result = conn.execute(stmt)
    print(f"ID inserido: {result.inserted_primary_key}")

    # Atualização condicional
    stmt = update(produtos).where(
        produtos.c.categoria == 'Informática'
    ).values(preco=produtos.c.preco * 1.1)  # 10% de aumento
    conn.execute(stmt)

    # Exclusão segura
    stmt = delete(produtos).where(
        produtos.c.preco < 10.0
    )
    linhas_afetadas = conn.execute(stmt).rowcount
    print(f"{linhas_afetadas} produtos removidos")

    conn.commit()  # Confirma as alterações

7. Junções e Subconsultas Avançadas

O Core suporta operações relacionais completas com joins, subconsultas e funções agregadas.

from sqlalchemy import func, exists

# Definindo tabela adicional para vendas
vendas = Table(
    'vendas',
    metadata,
    Column('id', Integer, primary_key=True),
    Column('produto_id', Integer, nullable=False),
    Column('quantidade', Integer),
    Column('data_venda', DateTime)
)
metadata.create_all(engine)

# JOIN entre tabelas
query = select(
    produtos.c.nome,
    func.sum(vendas.c.quantidade).label('total_vendido')
).join(
    vendas, produtos.c.id == vendas.c.produto_id
).group_by(produtos.c.nome)

# Subconsulta com EXISTS
subq = select(vendas.c.produto_id).where(
    vendas.c.data_venda >= '2024-01-01'
)
query = select(produtos).where(
    exists(subq.where(
        vendas.c.produto_id == produtos.c.id
    ))
)

# Funções agregadas
query = select(
    produtos.c.categoria,
    func.count().label('quantidade'),
    func.avg(produtos.c.preco).label('preco_medio'),
    func.sum(produtos.c.preco).label('valor_total')
).group_by(produtos.c.categoria)

8. Transações e Boas Práticas

Gerenciamento explícito de transações é crucial para consistência dos dados.

from sqlalchemy import text

# Transação manual
with engine.connect() as conn:
    trans = conn.begin()
    try:
        # Operações dentro da transação
        conn.execute(
            insert(produtos).values(nome='Mouse', preco=89.90)
        )

        # Savepoint para rollback parcial
        savepoint = conn.begin_nested()
        try:
            conn.execute(
                insert(produtos).values(nome='Teclado', preco=-50.00)
            )
        except Exception:
            savepoint.rollback()  # Rollback apenas desta operação
            print("Preço inválido, operação revertida")

        trans.commit()
    except Exception as e:
        trans.rollback()
        print(f"Transação revertida: {e}")

# Batch insert otimizado
with engine.connect() as conn:
    dados = [
        {'nome': 'Monitor', 'preco': 1200.00, 'categoria': 'Informática'},
        {'nome': 'Webcam', 'preco': 299.00, 'categoria': 'Periféricos'},
        {'nome': 'Headset', 'preco': 199.00, 'categoria': 'Áudio'},
    ]
    conn.execute(insert(produtos), dados)
    conn.commit()

Boas práticas adicionais:
- Use parâmetros nomeados em vez de concatenação de strings para evitar SQL injection
- Prefira with engine.connect() para garantir fechamento automático
- Utilize executemany() para inserções em lote com melhor performance
- Configure pool_size e max_overflow no Engine para ambientes de produção

O SQLAlchemy Core oferece uma camada de abstração poderosa que combina a flexibilidade do SQL puro com a segurança e produtividade de uma API Pythonica. Para cenários que exigem performance máxima ou controle granular sobre as consultas, é frequentemente superior ao ORM.

Referências