<h2>Introdução ao SQLAlchemy Core</h2>
<p>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.</p>
<p>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.</p>
<h2>Conexão com o Banco de Dados</h2>
<h3>Criando uma Engine</h3>
<p>A <code>Engine</code> é 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.</p>
<pre><code class="language-python">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')</code></pre>
<p>A string de conexão segue o padrão: <code>dialect+driver://username:password@host:port/database</code>. O parâmetro <code>echo=True</code> é ú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.</p>
<h3>Executando Queries Diretas</h3>
<p>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.</p>
<pre><code class="language-python">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,)</code></pre>
<p>O context manager (<code>with</code>) garante que a conexão seja liberada automaticamente. Se omitir o <code>text()</code>, o SQLAlchemy pode interpretar a string como um comando genérico, o que não é recomendado. O <code>text()</code> marca explicitamente que você quer SQL literal.</p>
<h2>Construindo Queries com Expressões</h2>
<h3>Usando Tabelas e Metadata</h3>
<p>No SQLAlchemy Core, você trabalha com objetos <code>Table</code> 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.</p>
<pre><code class="language-python">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)</code></pre>
<p>O <code>MetaData</code> é um container que armazena todas as definições de tabelas. Isso permite gerenciar múltiplas tabelas e suas relações. O <code>Column</code> define cada campo com seu tipo e constraints. Quando você chama <code>create_all()</code>, o SQLAlchemy gera o SQL <code>CREATE TABLE</code> apropriado para seu banco de dados específico.</p>
<h3>INSERT: Adicionando Dados</h3>
<p>Para inserir dados, usamos o método <code>insert()</code> da tabela, que cria um objeto de inserção SQL que pode ser customizado antes de executar.</p>
<pre><code class="language-python">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()</code></pre>
<p>O <code>insert()</code> retorna um objeto que ainda não executa nada — é apenas uma representação da query. O método <code>values()</code> define os dados a inserir. Quando você chama <code>execute()</code>, a query é finalmente enviada ao banco. O <code>commit()</code> confirma a transação; sem ele, os dados não são persistidos (apenas em autocommit mode).</p>
<h3>SELECT: Recuperando Dados</h3>
<p>As queries SELECT são a operação mais comum. O SQLAlchemy Core permite construir seleções complexas de forma expressiva e segura.</p>
<pre><code class="language-python">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}")</code></pre>
<p>Observe que <code>usuarios.c.nome</code> acessa a coluna <code>nome</code>. O atributo <code>.c</code> (columns) expõe todas as colunas da tabela. O SQLAlchemy constrói a query SQL correta para seu banco de dados. Operadores como <code>==</code>, <code>!=</code>, <code>.like()</code>, <code>.endswith()</code> são sobrecarregados para criar condições SQL automaticamente.</p>
<h3>UPDATE e DELETE</h3>
<p>Modificar e remover dados segue o mesmo padrão: criar um statement e executar.</p>
<pre><code class="language-python">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}")</code></pre>
<p>O <code>rowcount</code> 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 <code>where()</code> em updates e deletes — sem ela, você afeta todas as linhas da tabela.</p>
<h2>Transações e Gerenciamento de Conexões</h2>
<h3>Entendendo Transações ACID</h3>
<p>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.</p>
<p>No SQLAlchemy Core, você controla explicitamente quando fazer <code>commit()</code> (confirmar) ou <code>rollback()</code> (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.</p>
<pre><code class="language-python">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</code></pre>
<p>Se a validação falhar (saldo insuficiente), o <code>rollback()</code> desfaz tudo. Sem transações, a dedução teria acontecido mesmo sem a adição, deixando o banco inconsistente.</p>
<h3>Isolation Levels e Concorrência</h3>
<p>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.</p>
<pre><code class="language-python">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}")</code></pre>
<p>O <code>engine.begin()</code> é 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.</p>
<h2>Queries Avançadas</h2>
<h3>Joins e Relacionamentos</h3>
<p>Quando você trabalha com múltiplas tabelas relacionadas, precisa fazer joins. No SQLAlchemy Core, isso é declarativo e seguro.</p>
<pre><code class="language-python">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'}")</code></pre>
<p>O <code>join()</code> cria um INNER JOIN, enquanto <code>outerjoin()</code> 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.</p>
<h3>Agregações e GROUP BY</h3>
<p>Para relatórios e análises, você frequentemente precisa agrupar e agregar dados.</p>
<pre><code class="language-python">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}")</code></pre>
<p>O <code>func</code> é um namespace mágico que permite chamar qualquer função SQL suportada pelo banco de dados. O <code>.label()</code> renomeia o resultado para acessar facilmente: <code>row.total_posts</code> em vez de <code>row[0]</code>.</p>
<h3>Subqueries e CTEs</h3>
<p>Subqueries (consultas aninhadas) são poderosas para lógica complexa.</p>
<pre><code class="language-python">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")</code></pre>
<p>O <code>.alias()</code> 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.</p>
<h2>Práticas Recomendadas</h2>
<h3>Pool de Conexões</h3>
<p>O SQLAlchemy gerencia um pool de conexões por padrão, mas você deve conhecer como configurá-lo para produção.</p>
<pre><code class="language-python">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
)</code></pre>
<p>A configuração correta do pool evita esgotamento de conexões em produção. <code>pool_pre_ping=True</code> é essencial em ambientes instáveis onde conexões podem ser encerradas pelo servidor.</p>
<h3>Parametrização Segura</h3>
<p>Nunca concatene strings em SQL. O SQLAlchemy faz parametrização automaticamente com seus operadores, mas ao usar <code>text()</code>, você deve parametrizar manualmente.</p>
<pre><code class="language-python"># ✓ 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</code></pre>
<p>O SQLAlchemy Core previne SQL injection automaticamente quando você usa seus operadores. Apenas ao usar <code>text()</code> bruto você precisa ser cuidadoso e usar named parameters (<code>:nome</code>).</p>
<h2>Conclusão</h2>
<p>Neste artigo, você aprendeu que <strong>SQLAlchemy Core oferece controle fino sobre SQL mantendo segurança e portabilidade</strong>. Desde conexões com <code>create_engine()</code> até transações ACID com <code>commit()</code> e <code>rollback()</code>, você domina os fundamentos necessários para trabalhar com bancos de dados em Python de forma profissional.</p>
<p>Além disso, <strong>você conhece a diferença crítica entre construir queries programaticamente (com operadores e joins) versus SQL literal (com <code>text()</code>)</strong>: a primeira é mais segura e expressiva, enquanto a segunda é necessária apenas para queries muito específicas. Use-as estrategicamente.</p>
<p>Por fim, <strong>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</strong>. Aplique esses conhecimentos imediatamente em seus projetos e você verá a qualidade e a robustez subirem significativamente.</p>
<h2>Referências</h2>
<ul>
<li><a href="https://docs.sqlalchemy.org/en/20/core/" target="_blank" rel="noopener noreferrer">Documentação Oficial SQLAlchemy Core</a></li>
<li><a href="https://docs.sqlalchemy.org/en/20/tutorial/select.html" target="_blank" rel="noopener noreferrer">SQLAlchemy Core Tutorial - select() Statements</a></li>
<li><a href="https://docs.sqlalchemy.org/en/20/core/connections.html#using-transactions" target="_blank" rel="noopener noreferrer">Using Transactions with SQLAlchemy</a></li>
<li><a href="https://wiki.postgresql.org/wiki/Number_formatting" target="_blank" rel="noopener noreferrer">PostgreSQL + SQLAlchemy Best Practices</a></li>
<li><a href="https://www.oreilly.com/library/view/essential-sqlalchemy-2nd/9781491978528/" target="_blank" rel="noopener noreferrer">Essential SQLAlchemy 2e - Rick Copeland</a></li>
</ul>
<p><!-- FIM --></p>