O suporte asyncio do SQLAlchemy adapta Core e ORM a drivers assíncronos. A unidade de trabalho continua sendo uma sessão stateful: cada tarefa concorrente precisa da própria AsyncSession.

from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

engine = create_async_engine("postgresql+asyncpg://user:pass@db/app")
Session = async_sessionmaker(engine, expire_on_commit=False)


async def buscar_usuario(user_id: int):
    async with Session() as session:
        async with session.begin():
            return await session.scalar(
                select(Usuario).where(Usuario.id == user_id)
            )

Não versione credenciais; carregue a URL pelo mecanismo seguro da implantação. session.begin() delimita commit e rollback. Feche o engine no encerramento da aplicação com await engine.dispose().

Evite lazy loading que dispara I/O implícito fora de um await. Planeje relacionamentos com carregamento explícito e não execute chamadas bloqueantes dentro do loop. O guia de SQLAlchemy apresenta o ORM, e Alembic cobre migrações.

A documentação oficial de asyncio no SQLAlchemy, consultada em 22 de julho de 2026, detalha drivers, sessões e concorrência. Adote async quando a arquitetura inteira puder preservar I/O não bloqueante.

Uma sessão por requisição ou tarefa

Crie o engine uma vez na inicialização e compartilhe o async_sessionmaker. Abra uma AsyncSession nova para cada requisição, job em background ou tarefa concorrente. Fechar a sessão ao fim da unidade de trabalho faz parte do contrato: a conexão volta ao pool e o identity map é limpo.

Não guarde uma sessão em um serviço de longa duração e a reutilize entre awaits de tarefas diferentes. O uso concorrente de uma única sessão pode corromper a transação ou gerar erros difíceis de diagnosticar. Passe a sessão como dependência, ou abra-a na borda do caso de uso e injete repositórios que a recebem.

expire_on_commit=False é comum em apps web para que objetos continuem utilizáveis após o commit sem refresh imediato. Se precisar do estado fresco do banco, chame await session.refresh(obj) ou rode uma nova consulta. Prefira refresh explícito a lazy loads acidentais.

Transações e fronteiras de commit

async with session.begin() inicia a transação e faz commit em sucesso ou rollback em exceção. Trabalho aninhado deve usar a mesma sessão e a mesma transação, salvo quando você precisa de outra conexão de propósito. Evite misturar session.commit() espalhados no domínio com context managers externos que também fazem commit.

Quando um caso de uso abrange várias escritas que precisam ter sucesso juntas, mantenha-as no mesmo bloco begin(). Se um passo posterior falhar, os inserts anteriores devem reverter. Para consultas somente leitura, uma sessão curta sem transação de escrita basta, mas ainda assim feche a sessão com rapidez.

Trate erros de integridade na fronteira e mapeie-os para resultados de domínio como conflito ou validação. Não exponha texto específico do driver a clientes da API.

Carregue relacionamentos sem I/O implícito

O SQLAlchemy async desencoraja lazy loading porque o acesso a atributos não pode executar I/O com segurança. Carregue o necessário com selectinload, joinedload ou consultas explícitas antes de sair da sessão:

from sqlalchemy.orm import selectinload


async def buscar_pedido(session, pedido_id: int):
    stmt = (
        select(Pedido)
        .where(Pedido.id == pedido_id)
        .options(selectinload(Pedido.itens))
    )
    return await session.scalar(stmt)

Decida a estratégia de carregamento por caso de uso. Buscar grafos grandes demais desperdiça banda; buscar de menos força consultas extras ou acesso quebrado depois que a sessão fecha. Converta objetos ORM em schemas ou DTOs antes de devolver a resposta quando ela viver mais que a sessão.

Engine, pool e escolha de driver

create_async_engine precisa de um driver compatível com o dialeto, como asyncpg para PostgreSQL ou aiosqlite para testes locais com SQLite. Tamanho do pool, overflow e recycle continuam importando sob carga async: muitos checkouts concorrentes esperam ou falham, enquanto um pool exagerado pode sobrecarregar o banco.

Configure timeouts de statement e conexão conforme o driver e a implantação. Uma consulta travada sem timeout pode esgotar o pool mesmo com o event loop responsivo. Meça espera de checkout e duração de consulta em separado.

Não envolva um Session síncrono ou uma chamada bloqueante de DB-API em async def e chame isso no event loop. Se parte da pilha ainda for síncrona, rode em um worker thread ou mantenha esse caminho totalmente sync.

Testando repositórios async

Use o mesmo driver async que importa para testes de integração, ou um container com banco descartável. Um repositório fake em memória serve para regras de domínio, mas não prova SQL, constraints nem comportamento de transação.

import pytest


@pytest.fixture
async def session(engine):
    async with engine.connect() as connection:
        transaction = await connection.begin()
        Session = async_sessionmaker(bind=connection, expire_on_commit=False)
        async with Session() as session:
            yield session
        await transaction.rollback()

Confirme que as fronteiras de conexão e transação da fixture batem com a forma como a aplicação abre sessões. Código que cria um segundo engine ou faz commit em outra conexão pode escapar do rollback externo. Prefira schemas únicos ou bancos recriados quando o isolamento precisar ser mais forte.

Checklist operacional

Faça dispose do engine no encerramento. Mantenha credenciais fora do código e dos logs. Prefira carregamento explícito a acesso lazy. Limite a concorrência para que as tarefas não retirem mais conexões do que o pool consegue servir. Cubra sucesso, conflitos de integridade, resultados vazios e limpeza após cancelamento.

SQLAlchemy async ajuda quando o restante do caminho da requisição já é assíncrono. O ganho é I/O cooperativo, não velocidade mágica de consulta. Propriedade clara da sessão e transações explícitas importam mais do que converter todo def em async def.