SQLAlchemy assíncrono em APIs com banco relacional: erros que geram falhas, lentidão ou retrabalho

Como identificar problemas de sessão, transações, carregamento de dados e conexões em APIs com FastAPI e banco relacional.

Uma API de pedidos pode funcionar nos testes locais e apresentar erros HTTP 500, respostas demoradas ou gravações incompletas quando recebe requisições concorrentes. Com SQLAlchemy assíncrono, esses sintomas exigem investigar tanto o acesso ao banco quanto os limites das sessões e transações.

Este artigo acompanha os endpoints GET /pedidos e POST /pedidos para mostrar onde procurar evidências, quais padrões corrigir e como verificar o resultado. O objetivo é reduzir incidentes e retrabalho com decisões sustentadas por diagnóstico.

Referência de versões: os exemplos seguem a API do SQLAlchemy 2.0, com documentação da versão 2.0.54 consultada na redação, Python 3.10 ou superior e PostgreSQL via asyncpg. O exemplo de dependência com scope="function" requer FastAPI 0.121.0 ou superior. Os trechos são ilustrativos e pressupõem modelos ORM definidos; não representam uma aplicação executada ou um benchmark.

Como reconhecer os problemas antes de alterar a configuração

Antes de aumentar o pool ou trocar o driver, identifique em qual etapa a requisição perde tempo. A duração de uma resposta inclui aquisição de conexão, execução SQL, possíveis esperas por locks, processamento dos resultados e serialização.

Async permite atender outras tarefas enquanto uma operação aguarda I/O. Isso favorece concorrência, mas não garante que uma consulta individual termine mais rápido. Uma busca sem índice adequado continua custosa mesmo quando executada com await.

No GET /pedidos, registre a duração total e a quantidade de comandos SQL por requisição. Meça a aquisição de conexão separadamente: um evento de checkout informa quando a conexão foi entregue, mas sozinho não revela quanto tempo a tarefa aguardou.

O tempo observado na execução SQL também pode incluir espera por locks. Use a atividade do PostgreSQL para distinguir essa espera do processamento da consulta. Priorize gravações inconsistentes e falhas de produção antes de ajustes finos de latência.

Para complementar a configuração e os fundamentos, consulte o guia de implementação do SQLAlchemy assíncrono em APIs.

Misturar acesso síncrono e assíncrono pode bloquear a API

Declarar uma rota com async def não transforma chamadas bloqueantes em operações assíncronas. Se ela chama diretamente uma Session síncrona, a espera pelo banco pode bloquear o event loop e atrasar outras requisições daquele processo.

# Padrão problemático: chamada bloqueante dentro de async def.
async def listar_pedidos_incorreto(session_sync):
    return session_sync.execute(select(Pedido)).scalars().all()

Na configuração de SQLAlchemy async com FastAPI, engine e driver precisam ser compatíveis. Para SQLAlchemy com PostgreSQL via asyncpg, a URL utiliza o prefixo postgresql+asyncpg://.

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

engine = create_async_engine(database_url)
SessionFactory = async_sessionmaker(engine, expire_on_commit=False)

async def listar_pedidos(session):
    resultado = await session.execute(
        select(Pedido).order_by(Pedido.id).limit(50)
    )
    return resultado.scalars().all()

execute, flush e commit exigem await na sessão assíncrona. Já session.add(pedido) apenas registra o objeto na sessão e não recebe await. No resultado bufferizado de execute, scalars().all() é uma operação local; APIs de streaming têm outro comportamento.

Revise também chamadas HTTP síncronas e processamento pesado dentro das rotas. A adoção de um driver assíncrono resolve apenas a parte do fluxo que ele atende.

Compartilhar AsyncSession entre tarefas causa falhas de concorrência

A AsyncSession mantém estado mutável da operação. Compartilhar a mesma instância entre tarefas concorrentes permite que uma tarefa interfira na transação ou no estado usado pela outra.

A fábrica de sessões pode pertencer à aplicação. Cada tarefa concorrente deve obter sua própria sessão. Isso vale para consultas disparadas com asyncio.gather e para trabalhos executados em segundo plano.

import asyncio
from sqlalchemy import select

async def buscar_pedido(pedido_id):
    async with SessionFactory() as session:
        resultado = await session.execute(
            select(Pedido.id, Pedido.status)
            .where(Pedido.id == pedido_id)
        )
        linha = resultado.mappings().one_or_none()
        return dict(linha) if linha is not None else None

pedidos = await asyncio.gather(
    buscar_pedido(10),
    buscar_pedido(20),
)

Esse exemplo retorna dados materializados, evitando transportar objetos que ainda poderiam precisar da sessão. Trate também exceções das tarefas conforme o contrato da operação; uma falha em gather não representa rollback conjunto das demais sessões.

Consultas independentes podem usar sessões separadas. A criação de pedido e itens que precisa ser atômica deve ocorrer sequencialmente na mesma transação. Dividi-la entre sessões não cria uma confirmação conjunta.

Aumentar o número de tarefas também aumenta a disputa por conexões. Limite a concorrência quando o volume de operações superar a capacidade disponível.

MissingGreenlet: carregamento implícito durante a serialização

Um pedido carregado do banco pode conter atributos que ainda exigem SQL. Ao acessar pedido.itens durante a serialização, o ORM pode tentar fazer lazy loading em um ponto que não permite aguardar a operação.

Esse acesso é uma causa frequente de MissingGreenlet. Campos expirados ou adiados também merecem investigação. Se a sessão já foi encerrada, o mesmo desenho pode produzir um erro de objeto desconectado.

Defina quais dados a resposta precisa e carregue-os antes de construí-la:

from sqlalchemy import select
from sqlalchemy.orm import selectinload

async def consultar_pedido(session, pedido_id):
    resultado = await session.execute(
        select(Pedido)
        .where(Pedido.id == pedido_id)
        .options(selectinload(Pedido.itens))
    )
    pedido = resultado.scalar_one()
    return {
        "id": pedido.id,
        "itens": [
            {"id": item.id, "quantidade": item.quantidade}
            for item in pedido.itens
        ],
    }

A construção do DTO ocorre com os dados necessários carregados. Ao usar Pydantic, revise também os atributos acessados pelo modelo de resposta: a serialização não deve descobrir relacionamentos que precisam consultar o banco.

Para um objeto específico, await session.refresh(pedido, ["itens"]) permite solicitar o carregamento explicitamente. Modelos com AsyncAttrs oferecem await pedido.awaitable_attrs.itens quando esse acesso aguardável combina com o fluxo.

expire_on_commit=False evita a expiração automática após commit, mas não carrega relacionamentos ausentes nem mantém valores atualizados diante de alterações externas. Faça refresh quando a resposta exigir dados renovados.

Transações mal definidas geram gravações parciais e retrabalho

Considere um POST /pedidos que confirma o pedido e depois tenta inserir os itens. Se um item viola uma restrição, o pedido já confirmado permanece no banco. O cliente recebe uma falha e pode repetir a requisição, ampliando o problema.

Defina uma camada responsável pela transação da operação de negócio. Métodos auxiliares que fazem commit por conta própria dificultam preservar a atomicidade.

from sqlalchemy.exc import IntegrityError

async def criar_pedido(cliente_id, itens):
    try:
        async with SessionFactory.begin() as session:
            pedido = Pedido(cliente_id=cliente_id)
            session.add(pedido)
            await session.flush()

            for dados in itens:
                session.add(
                    ItemPedido(
                        pedido_id=pedido.id,
                        produto_id=dados["produto_id"],
                        quantidade=dados["quantidade"],
                    )
                )

            await session.flush()
            resposta = {"id": pedido.id}

        return resposta
    except IntegrityError:
        # A transação já foi revertida pelo contexto.
        # A camada HTTP deve classificar a restrição violada.
        raise

O flush envia alterações ao banco dentro da transação, permitindo obter o identificador e detectar restrições imediatas. Ele não confirma definitivamente a operação. O contexto confirma na saída bem-sucedida e reverte quando ocorre uma exceção, inclusive durante a confirmação.

Capture IntegrityError fora do contexto transacional, como no exemplo. Em controle manual, após falha de flush, execute await session.rollback() antes de reutilizar a sessão, ou encerre-a e descarte-a. Não trate toda violação como conflito de unicidade: examine a restrição e o contrato do endpoint.

Planeje o início da transação antes de consultas que possam iniciá-la implicitamente. Executar uma leitura e depois chamar session.begin() sem considerar o estado atual pode gerar conflito de controle transacional.

Quando o fluxo permitir, deixe chamadas externas demoradas fora da transação. Se a operação também precisa publicar eventos ou acionar outro serviço, defina uma estratégia de consistência; o commit no banco não confirma automaticamente ações externas.

Retentativas dependem do tipo de falha e da idempotência. Uma chave idempotente deve ter regra de unicidade e política para pedidos repetidos. Após uma desconexão durante commit, o resultado pode ser incerto: repetir cegamente pode duplicar a operação.

Sessões sem encerramento e engines por requisição desperdiçam recursos

Criar uma engine a cada requisição multiplica recursos e perde o benefício de reutilizar conexões. Mantenha a engine no ciclo de vida da aplicação, com recursos próprios em cada processo.

Uma dependência com yield delimita a sessão. O gerenciador de contexto garante sua limpeza na saída normal e quando o endpoint lança uma exceção.

import os
from contextlib import asynccontextmanager
from typing import Annotated
from fastapi import Depends, FastAPI, Request
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

@asynccontextmanager
async def lifespan(app: FastAPI):
    engine = create_async_engine(os.environ["DATABASE_URL"])
    app.state.session_factory = async_sessionmaker(
        engine, expire_on_commit=False
    )
    try:
        yield
    finally:
        await engine.dispose()

app = FastAPI(lifespan=lifespan)

async def get_session(request: Request):
    async with request.app.state.session_factory() as session:
        yield session

SessionDep = Annotated[
    AsyncSession,
    Depends(get_session, scope="function"),
]

No FastAPI 0.121.0 ou superior, scope="function" encerra a dependência depois da execução do endpoint e antes do envio da resposta. O escopo padrão de dependências com yield, request, mantém o recurso até o envio terminar.

O escopo de função combina com respostas construídas a partir de dados já materializados. Se um StreamingResponse precisa consultar o banco enquanto envia conteúdo, planeje uma sessão válida durante esse período e considere a retenção de conexão por clientes lentos.

A dependência do exemplo administra o recurso, sem definir commits automáticos. Confirme escritas dentro da operação para que falhas sejam tratadas antes de enviar uma resposta de sucesso.

Tarefas em segundo plano devem abrir suas próprias sessões e receber identificadores ou dados simples. Não prolongue artificialmente a sessão da requisição para atender trabalho que continuará depois dela.

Pool de conexões: aumentar limites sem diagnóstico pode piorar a lentidão

O pool de conexões SQLAlchemy limita recursos disponíveis para o acesso ao banco. Com PostgreSQL e a configuração usual de engine assíncrona, o SQLAlchemy utiliza AsyncAdaptedQueuePool, compatível com asyncio.

• pool_size: quantidade de conexões que o pool pode manter para reutilização.
• max_overflow: conexões adicionais permitidas quando a capacidade base está ocupada.
• pool_timeout: tempo de espera para obter uma conexão quando o limite foi atingido.

Em um cenário hipotético com quatro processos, cada um com pool_size=5 e max_overflow=5, o teto potencial é 4 × (5 + 5) = 40 conexões. Isso não significa que todas estarão abertas permanentemente. Réplicas, outras engines e clientes também entram no orçamento.

Antes de ampliar o limite, descubra por quanto tempo cada operação retém a conexão. Procure transações abertas durante chamadas externas, locks e consultas demoradas. Ampliar o pool sem capacidade no banco pode aumentar a contenção.

Timeout de pool é espera pela aquisição; timeout de consulta ocorre em outra etapa. pool_pre_ping ajuda a detectar conexões inválidas na retirada do pool, mas não recupera uma transação interrompida no meio da execução.

Defina os valores com medições de carga e margem para os demais consumidores do banco. Registre processos e réplicas junto à configuração para evitar multiplicações inesperadas.

Consultas N+1, falta de índices e excesso de dados continuam custando caro

Uma listagem que consulta pedidos e depois busca cliente e itens individualmente pode produzir consultas N+1. Conte os comandos SQL por requisição e observe como essa quantidade varia com o número de pedidos.

Em async, um lazy loading inadequado pode falhar antes mesmo de produzir todas essas consultas. Ao tornar os acessos aguardáveis, a execução pode funcionar e ainda continuar fazendo trabalho redundante.

selectinload costuma ser útil para coleções porque busca relacionamentos em consultas adicionais agrupadas. joinedload utiliza joins e pode repetir dados do registro principal quando há muitos filhos. A escolha depende da cardinalidade e do volume retornado.

from sqlalchemy.orm import joinedload

resultado = await session.execute(
    select(Pedido)
    .options(joinedload(Pedido.itens))
    .order_by(Pedido.id)
    .limit(50)
)
pedidos = resultado.unique().scalars().all()

Ao usar joinedload de coleções, unique() é necessário para deduplicar as entidades no resultado. Isso não elimina o custo das linhas transferidas pelo banco.

Para comparar estratégias no GET /pedidos, use o mesmo conjunto de dados e a mesma resposta. Registre quantidade de SQLs, duração, volume retornado e consumo de memória. A contagem exata com selectinload pode variar conforme os relacionamentos e o tamanho dos lotes.

Limite resultados, selecione os campos necessários e use ordenação determinística na paginação. Para volumes maiores, avalie paginação por cursor conforme os filtros e a navegação exigida.

No PostgreSQL, examine o plano antes de criar índices. EXPLAIN ANALYZE executa a consulta e mostra comportamento observado; planeje seu uso, especialmente em comandos que alteram dados. Async não elimina scans, bloqueios ou consultas repetidas.

Quando o cenário envolver SaaS e filtros por tenant, o conteúdo sobre índices parciais em SaaS com muitos tenants complementa a avaliação de índices adequados aos padrões de acesso.

Testes simplificados podem esconder falhas do banco usado em produção

SQLite é útil em determinados testes, mas não reproduz todos os comportamentos do PostgreSQL. Tipos, concorrência de escrita, bloqueios, restrições e recursos SQL podem variar. Um teste local aprovado não valida sozinho o caminho real de persistência.

Execute testes de integração com o mesmo tipo de banco e driver usados em produção, em ambiente isolado. Inclua dados sintéticos com variedade de pedidos, itens e filtros semelhante à que provoca os incidentes.

• Envie duas operações concorrentes com a mesma chave única, usando sessões distintas.
• Verifique o resultado previsto pelo contrato: conflito ou resposta idempotente, conforme a implementação.
• Confira que existe apenas o conjunto de registros permitido e que não ficaram pedidos parcialmente gravados.
• Provoque uma restrição em um item e confirme o rollback integral do pedido.
• Serialize respostas com relacionamentos para detectar atributos não carregados.
• Faça uma requisição posterior à falha para verificar que a aplicação continua atendendo.

No pipeline, aplique as migrações com Alembic em um banco limpo e execute os testes contra esse esquema. Valide também o caminho de atualização quando a mudança afetar instalações existentes.

Para reproduzir esgotamento do pool, configure um limite pequeno em um teste controlado e retenha conexões de forma conhecida. Verifique a resposta da aplicação e a recuperação após liberar os recursos. Isso ajuda a separar comportamento esperado de vazamentos.

Checklist para corrigir a API e comprovar o resultado

Comece pelo problema com maior impacto sobre os dados ou a disponibilidade. Para cada correção, registre a hipótese, a evidência que a sustenta e o resultado esperado.

1. Confirme a compatibilidade entre driver, engine e API assíncrona; revise os awaits.
2. Garanta uma AsyncSession por tarefa concorrente.
3. Carregue os atributos exigidos pela resposta antes da serialização.
4. Defina uma transação para cada operação que precisa ser atômica.
5. Verifique rollback e encerramento nos caminhos de exceção.
6. Revise lifespan, tarefas em segundo plano e duração de streams.
7. Some o teto de conexões de processos, réplicas e outros consumidores.
8. Conte consultas, examine planos e limite dados retornados.
9. Teste restrições, concorrência, migrações e respostas no banco representativo.
10. Compare resultados sob condições equivalentes e documente as decisões.

A latência p95 indica o tempo abaixo do qual ficam 95% das observações. Compare-a junto à taxa de erros, pois respostas rápidas que falham não demonstram melhoria. Mantenha iguais o volume de dados, a carga, a infraestrutura e os critérios de coleta.

Registre versões, configurações e decisões para que a equipe consiga reproduzir o cenário e detectar regressões. Melhorar a performance de APIs Python depende também de preservar a correção das operações durante a carga.

Escolha o endpoint que mais gera incidentes, aplique o checklist e compare os resultados antes e depois. Para aprofundar as próximas melhorias, explore os conteúdos de Python & FastAPI do blog Max Alex.