Como implementar SQLAlchemy assíncrono em APIs com banco relacional: guia prático para produção

Construir uma API concorrente não depende apenas de declarar rotas com async def. A camada de persistência precisa acompanhar esse modelo para evitar bloqueios no event loop, sessões compartilhadas e transações inconsistentes. Neste guia, você vai estruturar uma integração segura entre FastAPI, SQLAlchemy 2.x e PostgreSQL para uso em produção.

Quando usar SQLAlchemy assíncrono em uma API

O SQLAlchemy assíncrono é uma boa escolha quando a API atende várias requisições concorrentes e passa uma parte relevante do tempo aguardando operações de rede com o banco. Enquanto uma consulta está em andamento, o event loop pode atender outras tarefas compatíveis com async.

Isso não torna uma consulta lenta mais rápida por si só. Índices, modelagem, paginação e análise de plano de execução continuam sendo essenciais. A assincronicidade melhora o aproveitamento da concorrência de I/O; ela não substitui otimização de banco nem resolve tarefas intensivas de CPU.

Use esse modelo quando toda a cadeia crítica for assíncrona: rotas FastAPI, driver do banco, chamadas HTTP externas e clientes de fila, quando existirem. Para processamento pesado, como geração de relatórios extensos ou transformação de arquivos, prefira filas, workers ou processos separados.

Em uma API com muitas requisições simultâneas ao PostgreSQL, o driver asyncpg permite que a aplicação aguarde a resposta do banco sem bloquear outras corrotinas do mesmo processo.

Arquitetura recomendada para FastAPI, SQLAlchemy e PostgreSQL

Uma integração sustentável separa infraestrutura, persistência, regras de negócio e HTTP. A engine e a fábrica de sessões pertencem à infraestrutura; modelos ORM representam o banco; schemas Pydantic representam contratos de entrada e saída; serviços ou repositórios concentram operações de negócio; rotas apenas coordenam a requisição.

Crie uma única engine por processo da aplicação. Criar uma engine a cada requisição elimina os benefícios do pool de conexões e pode esgotar rapidamente o banco. Já a AsyncSession deve ter escopo curto: normalmente uma por requisição ou por unidade de trabalho.

app/
  api/          # rotas e dependências
  core/         # configuração e segurança
  db/           # engine, sessão e base ORM
  models/       # entidades SQLAlchemy
  schemas/      # contratos Pydantic
  services/     # regras de negócio
  migrations/   # revisões Alembic

Essa divisão reduz o acoplamento. Uma rota recebe a sessão por injeção de dependência, chama um serviço e devolve um schema de resposta, sem conter detalhes de SQL ou de transação.

Instalação das dependências e configuração da URL assíncrona

Para PostgreSQL, instale SQLAlchemy, o driver assíncrono e Alembic. Mantenha versões compatíveis e fixadas conforme a política do projeto.

pip install sqlalchemy asyncpg alembic

A URL deve usar o dialeto assíncrono:

DATABASE_URL=postgresql+asyncpg://usuario:senha@host:5432/meu_banco

Armazene essa configuração em variáveis de ambiente ou em um gerenciador de segredos. Credenciais não devem ser incluídas no código, em arquivos versionados ou em logs. Na inicialização, valide se a variável obrigatória existe e se possui um formato esperado.

Em ambientes com Docker, lembre-se de que o host da URL costuma ser o nome do serviço do banco na rede interna, e não necessariamente localhost.

Criando AsyncEngine e AsyncSession corretamente

O núcleo da integração é composto por uma AsyncEngine compartilhada e uma fábrica async_sessionmaker. A configuração do pool deve respeitar a capacidade do PostgreSQL, o número de workers e a quantidade de réplicas da aplicação.

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

DATABASE_URL = "postgresql+asyncpg://usuario:senha@host:5432/meu_banco"

engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,
    echo=False,
)

SessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

pool_pre_ping=True ajuda a detectar conexões obsoletas antes do uso. Já expire_on_commit=False evita que atributos já carregados precisem ser buscados novamente após um commit, o que reduz carregamentos implícitos inesperados.

Não escolha valores altos de pool por padrão. O total potencial de conexões é influenciado por pool_size + max_overflow, multiplicado pelos processos e réplicas em execução. Esse total precisa caber no limite do banco, com margem para migrations, administração e outros serviços.

Injetando a sessão por requisição no FastAPI

Uma dependência com yield fornece uma sessão à rota e garante o encerramento ao fim do ciclo da requisição. A sessão não deve ser armazenada globalmente nem reutilizada simultaneamente por várias requisições.

from collections.abc import AsyncGenerator
from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI()

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with SessionLocal() as session:
        yield session

@app.get("/health/db")
async def database_health(db: AsyncSession = Depends(get_db)):
    return {"status": "ok"}

A dependência cuida do ciclo de vida da sessão, mas a política de commit precisa ser explícita. Para operações de escrita, mantenha o commit dentro do serviço ou de uma unidade de trabalho bem definida. Isso evita commits automáticos em locais difíceis de auditar.

Se uma exceção ocorrer depois de uma falha de banco, execute rollback antes de reutilizar a sessão. Em muitos casos, o contexto de transação com session.begin() torna esse fluxo mais claro e seguro.

Modelando entidades e executando consultas assíncronas

Com SQLAlchemy 2.x, prefira a API moderna baseada em select(). Modelos tipados tornam as relações e colunas mais legíveis, enquanto consultas explícitas facilitam paginação, filtros e revisão de desempenho.

from sqlalchemy import Boolean, String, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    active: Mapped[bool] = mapped_column(Boolean, default=True)

async def list_active_users(session: AsyncSession) -> list[User]:
    statement = select(User).where(User.active.is_(True))
    result = await session.scalars(statement)
    return list(result.all())

Use session.scalars() quando o resultado esperado for uma coleção de entidades. Para uma entidade por chave primária, await session.get(User, user_id) é direto. Em operações de criação, use add(), flush() quando precisar do identificador antes do commit e refresh() quando for necessário recarregar valores produzidos pelo banco.

Listagens devem paginar no banco. Evite carregar toda uma tabela para depois cortar a lista em memória. Também aplique filtros e ordenação determinística na consulta para resultados previsíveis.

Transações, commit, rollback e tratamento de erros

Alterações relacionadas precisam pertencer à mesma transação. O contexto session.begin() confirma as mudanças ao terminar sem erro e executa rollback se uma exceção interromper a operação.

from fastapi import HTTPException
from sqlalchemy.exc import IntegrityError

async def create_user(session: AsyncSession, email: str) -> User:
    user = User(email=email)
    try:
        async with session.begin():
            session.add(user)
            await session.flush()
        await session.refresh(user)
        return user
    except IntegrityError as exc:
        raise HTTPException(
            status_code=409,
            detail="Já existe um usuário com este e-mail.",
        ) from exc

Em um handler mais amplo, faça rollback quando necessário antes de mapear a falha para um erro de domínio. Não devolva ao cliente mensagens brutas do driver, nomes de tabelas, SQL ou dados de conexão. Registre os detalhes técnicos de forma segura nos logs e retorne uma mensagem adequada ao contrato da API.

Também diferencie falhas de validação das falhas de persistência: um campo ausente ou inválido deve resultar em resposta de validação; uma violação de unicidade ou de chave estrangeira deve resultar em uma resposta de conflito ou regra de negócio.

Relacionamentos e o problema do carregamento preguiçoso no async

O lazy loading pode causar problemas em aplicações assíncronas porque acessar um relacionamento aparentemente simples pode disparar I/O implícito fora do contexto esperado. Um sintoma comum é o erro MissingGreenlet, especialmente durante a serialização da resposta.

Carregue antecipadamente os relacionamentos que a resposta realmente precisa. Para coleções, selectinload costuma ser uma opção segura; para casos específicos, joinedload pode reduzir consultas, desde que o volume de linhas produzido pelo join seja avaliado.

from sqlalchemy import select
from sqlalchemy.orm import selectinload

statement = (
    select(Order)
    .options(selectinload(Order.items))
    .where(Order.id == order_id)
)
order = (await session.scalars(statement)).one_or_none()

Evite serializar entidades ORM diretamente depois que a sessão foi encerrada. Converta os dados para schemas de resposta enquanto os atributos necessários já estão carregados. Isso torna a resposta previsível e elimina consultas acidentais durante a renderização.

Migrations com Alembic em projetos assíncronos

O Alembic versiona a evolução do schema. Cada alteração estrutural deve gerar uma revisão rastreável, revisada em pull request e validada antes de chegar à produção.

alembic init migrations
alembic revision --autogenerate -m "add users table"
alembic upgrade head

O autogenerate é um ponto de partida, não uma garantia de que a migration está correta. Revise índices, constraints, valores padrão, operações destrutivas e mudanças de tipo. Um arquivo de migration deve ser seguro para o ambiente e refletir a estratégia de deploy.

Mesmo em uma aplicação que usa SQLAlchemy assíncrono em runtime, o fluxo de migrations pode ser executado de forma controlada pelo ambiente do Alembic. O aspecto importante é apontar o target_metadata para os metadados dos modelos e configurar corretamente a URL de conexão do ambiente.

Não aplique migrations automaticamente na inicialização de cada réplica. Execute-as uma vez no pipeline ou em uma etapa de deploy com controle de concorrência. Para mudanças grandes, planeje etapas compatíveis entre versões da aplicação para reduzir risco de indisponibilidade.

Boas práticas de produção: performance, testes e observabilidade

Produção exige mais do que código funcional. Defina timeouts de conexão e de consulta de acordo com o comportamento do serviço, limite o pool de conexões e monitore erros, latência, espera pelo pool e consultas lentas.

Revise índices a partir das consultas reais da aplicação. Uma coluna usada com frequência em filtros, joins ou ordenações pode exigir índice, mas cada índice adicional também aumenta o custo de escrita. Use métricas e planos de execução para decidir, em vez de aplicar índices indiscriminadamente.

Os testes de integração devem usar um PostgreSQL isolado e aplicar migrations antes da suíte. Cubra criação, leitura, concorrência relevante, conflitos de unicidade, rollback e relacionamentos. Mocks são úteis nas bordas, mas não substituem a validação do SQL, do driver e das constraints no banco real.

Adote logs estruturados com identificador de requisição e sinais técnicos suficientes para investigação, sem registrar senhas ou dados sensíveis. Métricas de duração de consultas, taxa de erros e saturação do pool ajudam a identificar degradações antes de elas se transformarem em indisponibilidade.

Checklist final para colocar a API em produção

• A URL do banco e os segredos estão fora do repositório e são validados na inicialização.
• Existe uma única engine por processo e uma AsyncSession isolada por requisição ou unidade de trabalho.
• O pool foi dimensionado considerando workers, réplicas e o limite de conexões do PostgreSQL.
• Operações de escrita usam transações explícitas, rollback e respostas de erro sem detalhes internos.
• Consultas usam paginação, filtros no banco e carregamento explícito dos relacionamentos necessários.
• Migrations Alembic estão versionadas, revisadas e são aplicadas pelo pipeline de deploy.
• Testes de integração cobrem os fluxos críticos com um banco PostgreSQL real e isolado.
• Logs, métricas, alertas e acompanhamento de consultas lentas estão configurados.

Com esses pontos atendidos, a API terá uma base mais previsível para crescer. O SQLAlchemy assíncrono funciona melhor quando faz parte de uma arquitetura deliberada: sessões curtas, transações claras, consultas observáveis e limites compatíveis com o banco.

Sua empresa precisa de uma API FastAPI robusta, escalável e pronta para integrar dados com segurança? Entre em contato com a Max Alex para planejar e implementar uma arquitetura Python adequada ao seu produto.