Introdução ao SQLAlchemy Core
SQLAlchemy é a biblioteca SQL mais madura e robusta do ecossistema Python. Diferentemente do SQLAlchemy ORM, que abstrai completamente o banco de dados através de mapeamento de classes, o SQLAlchemy Core oferece uma interface mais direta e controlável com a linguagem SQL, mantendo a flexibilidade e o poder expressivo das queries.
O Core é ideal quando você precisa de controle fino sobre as operações SQL, trabalha com bancos de dados complexos, ou simplesmente prefere escrever SQL sem abstrações excessivas. Neste artigo, você aprenderá desde a conexão com o banco até transações avançadas, sempre com exemplos práticos que funcionam imediatamente.
Conexão com o Banco de Dados
Criando uma Engine
A Engine é o ponto de entrada para todas as operações no SQLAlchemy Core. Ela gerencia um pool de conexões e encapsula a lógica de comunicação com o banco de dados. A engine não abre conexões imediatamente — ela as cria sob demanda, otimizando recursos.
from sqlalchemy import create_engine
# PostgreSQL
engine = create_engine(
'postgresql://usuario:senha@localhost:5432/meu_banco',
echo=True # Mostra as queries SQL executadas (apenas para desenvolvimento)
)
# SQLite (perfeito para aprendizado)
engine = create_engine('sqlite:///banco.db')
# MySQL
engine = create_engine('mysql+pymysql://usuario:senha@localhost:3306/meu_banco')
A string de conexão segue o padrão: dialect+driver://username:password@host:port/database. O parâmetro echo=True é útil durante o desenvolvimento, mas deve ser removido em produção. O SQLAlchemy usa um pool de conexões por padrão, reutilizando conexões para melhor performance.
Executando Queries Diretas
Uma vez com a engine, você pode executar queries SQL brutas. Isso é útil para operações simples ou quando precisa de SQL muito específico que não vale a pena mapear com construções do Core.
from sqlalchemy import text
engine = create_engine('sqlite:///exemplo.db')
# Executar uma query simples
with engine.connect() as connection:
resultado = connection.execute(text('SELECT 1 as numero'))
for row in resultado:
print(row) # (1,)
O context manager (with) garante que a conexão seja liberada automaticamente. Se omitir o text(), o SQLAlchemy pode interpretar a string como um comando genérico, o que não é recomendado. O text() marca explicitamente que você quer SQL literal.
Construindo Queries com Expressões
Usando Tabelas e Metadata
No SQLAlchemy Core, você trabalha com objetos Table que representam tabelas no banco de dados. Esses objetos permitem construir queries de forma programática, segura contra SQL injection e com validação em tempo de execução.
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, DateTime
from datetime import datetime
engine = create_engine('sqlite:///exemplo.db')
metadata = MetaData()
# Definir uma tabela
usuarios = Table(
'usuarios',
metadata,
Column('id', Integer, primary_key=True),
Column('nome', String(100), nullable=False),
Column('email', String(100), unique=True),
Column('criado_em', DateTime, default=datetime.utcnow)
)
# Criar a tabela no banco (se não existir)
metadata.create_all(engine)
O MetaData é um container que armazena todas as definições de tabelas. Isso permite gerenciar múltiplas tabelas e suas relações. O Column define cada campo com seu tipo e constraints. Quando você chama create_all(), o SQLAlchemy gera o SQL CREATE TABLE apropriado para seu banco de dados específico.
INSERT: Adicionando Dados
Para inserir dados, usamos o método insert() da tabela, que cria um objeto de inserção SQL que pode ser customizado antes de executar.
from sqlalchemy import insert
# Forma 1: Insert simples com uma linha
stmt = insert(usuarios).values(nome='João Silva', email='joao@example.com')
with engine.connect() as connection:
resultado = connection.execute(stmt)
connection.commit() # Confirma a transação
print(f"ID da linha inserida: {resultado.inserted_primary_key}")
# Forma 2: Inserir múltiplas linhas
dados = [
{'nome': 'Maria Santos', 'email': 'maria@example.com'},
{'nome': 'Pedro Costa', 'email': 'pedro@example.com'},
{'nome': 'Ana Oliveira', 'email': 'ana@example.com'},
]
stmt = insert(usuarios)
with engine.connect() as connection:
connection.execute(stmt, dados)
connection.commit()
O insert() retorna um objeto que ainda não executa nada — é apenas uma representação da query. O método values() define os dados a inserir. Quando você chama execute(), a query é finalmente enviada ao banco. O commit() confirma a transação; sem ele, os dados não são persistidos (apenas em autocommit mode).
SELECT: Recuperando Dados
As queries SELECT são a operação mais comum. O SQLAlchemy Core permite construir seleções complexas de forma expressiva e segura.
from sqlalchemy import select, and_, or_
# SELECT simples
stmt = select(usuarios)
with engine.connect() as connection:
resultado = connection.execute(stmt)
for row in resultado:
print(row) # Row(id=1, nome='João Silva', email='joao@example.com', criado_em=...)
# SELECT com WHERE
stmt = select(usuarios).where(usuarios.c.nome == 'João Silva')
with engine.connect() as connection:
row = connection.execute(stmt).first()
if row:
print(f"Nome: {row.nome}, Email: {row.email}")
# SELECT com AND/OR
stmt = select(usuarios).where(
and_(
usuarios.c.nome.like('%Silva'),
usuarios.c.email.endswith('@example.com')
)
)
with engine.connect() as connection:
for row in connection.execute(stmt):
print(f"{row.nome} - {row.email}")
# SELECT apenas colunas específicas
stmt = select(usuarios.c.nome, usuarios.c.email)
with engine.connect() as connection:
for row in connection.execute(stmt):
print(f"{row.nome}: {row.email}")
Observe que usuarios.c.nome acessa a coluna nome. O atributo .c (columns) expõe todas as colunas da tabela. O SQLAlchemy constrói a query SQL correta para seu banco de dados. Operadores como ==, !=, .like(), .endswith() são sobrecarregados para criar condições SQL automaticamente.
UPDATE e DELETE
Modificar e remover dados segue o mesmo padrão: criar um statement e executar.
from sqlalchemy import update, delete
# UPDATE
stmt = update(usuarios).where(usuarios.c.id == 1).values(nome='João Atualizado')
with engine.connect() as connection:
connection.execute(stmt)
connection.commit()
# DELETE
stmt = delete(usuarios).where(usuarios.c.email.like('%old_domain%'))
with engine.connect() as connection:
resultado = connection.execute(stmt)
connection.commit()
print(f"Linhas deletadas: {resultado.rowcount}")
O rowcount retorna quantas linhas foram afetadas pela operação. Isso é útil para validar se a operação teve o efeito esperado. Sempre use uma cláusula where() em updates e deletes — sem ela, você afeta todas as linhas da tabela.
Transações e Gerenciamento de Conexões
Entendendo Transações ACID
Uma transação é um conjunto de operações SQL que deve ser executado atomicamente: ou todas as operações succedem, ou nenhuma é confirmada no banco. SQLAlchemy implementa o modelo ACID (Atomicidade, Consistência, Isolamento, Durabilidade) nativamente.
No SQLAlchemy Core, você controla explicitamente quando fazer commit() (confirmar) ou rollback() (desfazer). Sem um commit, as mudanças permanecem em memória e são perdidas quando a conexão fecha. Isso oferece segurança: você só persistir dados quando tem certeza de que tudo está correto.
from sqlalchemy import insert, select
# Cenário: Transferência de saldo entre contas
contas = Table(
'contas',
metadata,
Column('id', Integer, primary_key=True),
Column('titular', String(100)),
Column('saldo', Integer) # em centavos para evitar problemas com float
)
metadata.create_all(engine)
# Inserir dados iniciais
with engine.connect() as conn:
conn.execute(insert(contas).values(titular='Alice', saldo=100000))
conn.execute(insert(contas).values(titular='Bob', saldo=50000))
conn.commit()
# Transferência: se falhar no meio, tudo volta
def transferir(origem_id, destino_id, valor):
with engine.connect() as conn:
try:
# Deduzir de origem
conn.execute(
update(contas)
.where(contas.c.id == origem_id)
.values(saldo=contas.c.saldo - valor)
)
# Validação: rejeitar se saldo ficar negativo
resultado = conn.execute(
select(contas.c.saldo).where(contas.c.id == origem_id)
).scalar()
if resultado < 0:
raise ValueError("Saldo insuficiente")
# Adicionar ao destino
conn.execute(
update(contas)
.where(contas.c.id == destino_id)
.values(saldo=contas.c.saldo + valor)
)
conn.commit()
print("Transferência realizada com sucesso")
except Exception as e:
conn.rollback()
print(f"Erro na transferência: {e}")
transferir(1, 2, 30000) # Transferir 300,00 de Alice para Bob
Se a validação falhar (saldo insuficiente), o rollback() desfaz tudo. Sem transações, a dedução teria acontecido mesmo sem a adição, deixando o banco inconsistente.
Isolation Levels e Concorrência
Quando múltiplas conexões acessam o banco simultaneamente, há risco de condições de corrida. O SQLAlchemy permite controlar o nível de isolamento da transação.
from sqlalchemy import event
# Usar isolation_level específico (PostgreSQL)
engine = create_engine(
'postgresql://user:pass@localhost/db',
isolation_level='SERIALIZABLE' # Mais restritivo, mais seguro
)
# Níveis comuns:
# - READ UNCOMMITTED: lê dados não confirmados (raro)
# - READ COMMITTED: lê apenas dados confirmados (default em PostgreSQL)
# - REPEATABLE READ: garante que dados lidos não mudem na transação
# - SERIALIZABLE: simula execução sequencial (mais lento, mais seguro)
with engine.begin() as connection: # begin() auto-faz commit ao sair, rollback em erro
resultado = connection.execute(
select(contas.c.saldo).where(contas.c.id == 1)
).scalar()
print(f"Saldo atual: {resultado}")
O engine.begin() é uma variação que automatiza commit/rollback. Se nenhuma exceção ocorrer, faz commit automaticamente. Se uma exceção for levantada, faz rollback. Isso reduz boilerplate para casos simples.
Queries Avançadas
Joins e Relacionamentos
Quando você trabalha com múltiplas tabelas relacionadas, precisa fazer joins. No SQLAlchemy Core, isso é declarativo e seguro.
from sqlalchemy import ForeignKey, select, join
# Definir tabelas com relacionamento
usuarios_tabela = Table(
'usuarios',
metadata,
Column('id', Integer, primary_key=True),
Column('nome', String(100))
)
posts = Table(
'posts',
metadata,
Column('id', Integer, primary_key=True),
Column('usuario_id', Integer, ForeignKey('usuarios.id')),
Column('titulo', String(200)),
Column('conteudo', String(1000))
)
metadata.create_all(engine)
# Inserir dados de exemplo
with engine.begin() as conn:
conn.execute(insert(usuarios_tabela).values(nome='Alice'))
conn.execute(insert(usuarios_tabela).values(nome='Bob'))
conn.execute(insert(posts).values(usuario_id=1, titulo='Post 1', conteudo='Conteúdo'))
conn.execute(insert(posts).values(usuario_id=1, titulo='Post 2', conteudo='Mais conteúdo'))
conn.execute(insert(posts).values(usuario_id=2, titulo='Post 3', conteudo='Outro post'))
# INNER JOIN: retorna apenas linhas que têm match em ambas as tabelas
stmt = (
select(usuarios_tabela.c.nome, posts.c.titulo)
.select_from(
join(usuarios_tabela, posts, usuarios_tabela.c.id == posts.c.usuario_id)
)
)
with engine.connect() as conn:
for row in conn.execute(stmt):
print(f"{row.nome} escreveu: {row.titulo}")
# LEFT JOIN: retorna todos os usuários, mesmo sem posts
from sqlalchemy import outerjoin
stmt = (
select(usuarios_tabela.c.nome, posts.c.titulo)
.select_from(
outerjoin(usuarios_tabela, posts, usuarios_tabela.c.id == posts.c.usuario_id)
)
)
with engine.connect() as conn:
for row in conn.execute(stmt):
nome, titulo = row.nome, row.titulo
print(f"{nome}: {titulo or 'Nenhum post'}")
O join() cria um INNER JOIN, enquanto outerjoin() cria um LEFT OUTER JOIN. O segundo argumento do join é a condição de junção. O SQLAlchemy é inteligente o suficiente para inferir muitas junções automaticamente baseado em ForeignKey.
Agregações e GROUP BY
Para relatórios e análises, você frequentemente precisa agrupar e agregar dados.
from sqlalchemy import func, group_by
# Contar quantos posts cada usuário tem
stmt = (
select(
usuarios_tabela.c.nome,
func.count(posts.c.id).label('total_posts')
)
.select_from(
outerjoin(usuarios_tabela, posts, usuarios_tabela.c.id == posts.c.usuario_id)
)
.group_by(usuarios_tabela.c.id, usuarios_tabela.c.nome)
)
with engine.connect() as conn:
for row in conn.execute(stmt):
print(f"{row.nome}: {row.total_posts} posts")
# Outros agregadores: func.sum(), func.avg(), func.max(), func.min()
stmt = (
select(
usuarios_tabela.c.nome,
func.count(posts.c.id).label('posts'),
func.max(posts.c.titulo).label('ultimo_titulo')
)
.select_from(
outerjoin(usuarios_tabela, posts, usuarios_tabela.c.id == posts.c.usuario_id)
)
.group_by(usuarios_tabela.c.id)
)
with engine.connect() as conn:
for row in conn.execute(stmt):
print(f"{row.nome}: {row.posts} posts, último: {row.ultimo_titulo}")
O func é um namespace mágico que permite chamar qualquer função SQL suportada pelo banco de dados. O .label() renomeia o resultado para acessar facilmente: row.total_posts em vez de row[0].
Subqueries e CTEs
Subqueries (consultas aninhadas) são poderosas para lógica complexa.
from sqlalchemy import literal_column
# Subquery: usuários que têm mais de 1 post
posts_por_usuario = (
select(
posts.c.usuario_id,
func.count(posts.c.id).label('total')
)
.group_by(posts.c.usuario_id)
.having(func.count(posts.c.id) > 1)
.alias('posts_subquery')
)
stmt = (
select(usuarios_tabela.c.nome, posts_por_usuario.c.total)
.select_from(
join(usuarios_tabela, posts_por_usuario,
usuarios_tabela.c.id == posts_por_usuario.c.usuario_id)
)
)
with engine.connect() as conn:
for row in conn.execute(stmt):
print(f"{row.nome}: {row.total} posts")
O .alias() converte a subquery em algo que pode ser usado em joins e selects subsequentes. Isso é útil quando a lógica é complexa demais para uma única query.
Práticas Recomendadas
Pool de Conexões
O SQLAlchemy gerencia um pool de conexões por padrão, mas você deve conhecer como configurá-lo para produção.
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool, NullPool
# Pool padrão (QueuePool): reutiliza conexões
engine = create_engine(
'postgresql://user:pass@localhost/db',
poolclass=QueuePool,
pool_size=10, # Manter 10 conexões abertas
max_overflow=20, # Permitir até 20 conexões extras se necessário
pool_recycle=3600, # Reciclar conexões a cada hora (útil para conexões que timeout)
pool_pre_ping=True # Testar conexão antes de usar (evita "connection lost")
)
# NullPool: não reutiliza (útil para serverless/Lambda)
engine_serverless = create_engine(
'postgresql://user:pass@localhost/db',
poolclass=NullPool
)
A configuração correta do pool evita esgotamento de conexões em produção. pool_pre_ping=True é essencial em ambientes instáveis onde conexões podem ser encerradas pelo servidor.
Parametrização Segura
Nunca concatene strings em SQL. O SQLAlchemy faz parametrização automaticamente com seus operadores, mas ao usar text(), você deve parametrizar manualmente.
# ✓ SEGURO: SQLAlchemy cuida da parametrização
stmt = select(usuarios).where(usuarios.c.nome == 'João')
# ✓ SEGURO: Parametrização explícita com text()
stmt = select(usuarios).where(usuarios.c.nome == bindparam('nome'))
with engine.connect() as conn:
resultado = conn.execute(stmt, {'nome': 'João'})
# ✗ INSEGURO: SQL Injection!
nome = "' OR '1'='1"
stmt = text(f"SELECT * FROM usuarios WHERE nome = '{nome}'") # Perigoso!
# ✓ CORRETO com text():
stmt = text("SELECT * FROM usuarios WHERE nome = :nome")
with engine.connect() as conn:
resultado = conn.execute(stmt, {'nome': "' OR '1'='1"}) # Protegido
O SQLAlchemy Core previne SQL injection automaticamente quando você usa seus operadores. Apenas ao usar text() bruto você precisa ser cuidadoso e usar named parameters (:nome).
Conclusão
Neste artigo, você aprendeu que SQLAlchemy Core oferece controle fino sobre SQL mantendo segurança e portabilidade. Desde conexões com create_engine() até transações ACID com commit() e rollback(), você domina os fundamentos necessários para trabalhar com bancos de dados em Python de forma profissional.
Além disso, você conhece a diferença crítica entre construir queries programaticamente (com operadores e joins) versus SQL literal (com text()): a primeira é mais segura e expressiva, enquanto a segunda é necessária apenas para queries muito específicas. Use-as estrategicamente.
Por fim, lembre-se que configuração e boas práticas em produção (pool de conexões, parametrização, níveis de isolamento) separar código amador de código profissional. Aplique esses conhecimentos imediatamente em seus projetos e você verá a qualidade e a robustez subirem significativamente.