Transações no SQLAlchemy: Session, Commit e Rollback

Transações no SQLAlchemy mantêm um conjunto de alterações como uma única unidade: ou todas são confirmadas com commit, ou todas são desfeitas com rollback. A Session acompanha objetos, executa o padrão unit of work e obtém uma conexão quando precisa falar com o banco.

Antes deste tema, vale revisar o guia completo de SQLAlchemy. O ponto central é separar o escopo da transação do tempo de vida arbitrário da aplicação.

Use context manager para delimitar a transação

from sqlalchemy import create_engine
from sqlalchemy.orm import Session

engine = create_engine("postgresql+psycopg://app:senha@localhost/app")

def transferir(origem, destino, valor):
    with Session(engine) as session:
        with session.begin():
            origem.saldo -= valor
            destino.saldo += valor
            session.add_all([origem, destino])

Se o bloco terminar normalmente, session.begin() confirma. Se surgir uma exceção, ele reverte e propaga o erro. O bloco externo fecha a Session e devolve recursos ao pool. Não capture uma exceção apenas para ignorá-la, pois o chamador precisa saber que a transferência falhou.

Também é possível criar uma fábrica:

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)

with SessionLocal.begin() as session:
    session.add(novo_pedido)

flush não é commit

flush() envia INSERT, UPDATE e DELETE pendentes ao banco, mas continua dentro da transação. Ele é útil para obter uma chave gerada antes de criar outro objeto:

with SessionLocal.begin() as session:
    pedido = Pedido(cliente_id=42)
    session.add(pedido)
    session.flush()
    session.add(Item(pedido_id=pedido.id, sku="PY-01"))

Uma consulta ou commit() pode provocar flush automático. Se uma restrição falhar durante o flush, a Session fica inativa para aquela transação. Faça rollback() ou deixe o context manager cuidar disso antes de continuar.

Escopo de Session em aplicações web

Uma prática comum é abrir uma Session por requisição, executar a regra de negócio, confirmar operações de escrita e fechar ao final. Não compartilhe a mesma instância entre threads ou tarefas assíncronas. Para código assíncrono, use uma AsyncSession por tarefa e consulte o guia de SQLAlchemy assíncrono.

Evite chamar commit() dentro de cada função de repositório. Isso impede que uma operação de negócio com várias etapas seja atômica. Prefira deixar a camada que conhece a unidade de trabalho decidir quando confirmar.

Savepoint para recuperar apenas uma etapa

begin_nested() cria um savepoint quando o banco oferece suporte:

with SessionLocal.begin() as session:
    session.add(lote)
    try:
        with session.begin_nested():
            session.add(item_opcional)
            session.flush()
    except IntegrityError:
        registrar_item_rejeitado()

O erro reverte o savepoint, não necessariamente a transação externa. Use esse recurso com intenção clara; ele não substitui a definição correta dos limites da transação.

Testes e erros comuns

Teste sucesso e falha no meio da operação, verificando que não restou estado parcial. Evite Session global, transações abertas durante chamadas HTTP lentas e tratamento genérico que oculta erros. Mantenha a transação curta, mas longa o suficiente para representar uma regra de negócio indivisível.

A documentação oficial de transações do SQLAlchemy, consultada em 28 de julho de 2026, detalha autobegin, context managers, savepoints e níveis de isolamento.