Rotas Assíncronas e Bancos de Dados
Combine rotas async def com sessões assíncronas SQLAlchemy 2.0 e asyncpg.
Receita
Cartão de referência rápida - pronto para copiar e colar.
from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy import text
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db", pool_pre_ping=True)
Session = async_sessionmaker(engine, expire_on_commit=False)
app = FastAPI()
async def get_session():
async with Session() as s:
yield s
@app.get("/ping")
async def ping(session: AsyncSession = Depends(get_session)):
await session.execute(text("SELECT 1"))
return {"db": "ok"}Quando usar isso:
- APIs de I/O de alta concorrência
- PostgreSQL com asyncpg
- Chamadas upstream HTTP não bloqueantes
Exemplo Funcional
from sqlalchemy import select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi import Depends, FastAPI
class Base(DeclarativeBase):
pass
class Item(Base):
__tablename__ = "items"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column()
# assuma Session/get_session da receita
app = FastAPI()
@app.get("/items")
async def list_items(session: AsyncSession = Depends(get_session)) -> list[dict]:
rows = (await session.execute(select(Item).limit(20))).scalars().all()
return [{"id": r.id, "name": r.name} for r in rows]O que isso demonstra:
- Consultas ORM assíncronas
- Sessões com escopo de requisição
- Estilo de seleção SQLAlchemy 2.0
Mergulho Profundo
Como Funciona
- Uvicorn agenda corrotinas em um loop de eventos.
- Motores assíncronos usam asyncpg para PostgreSQL.
- Mantenha transações curtas por requisição.
Armadilhas
- Validação de limite ignorada - Dados inválidos chegam às camadas de persistência. Correção: Valide com Pydantic ou formulários do framework na borda.
- Vazamento de rastros de pilha - Clientes veem erros internos. Correção: Mapeie exceções para respostas HTTP estáveis.
- Bloqueio de loops de eventos assíncronos - Workers param sob carga concorrente. Correção: Use drivers assíncronos ou wrappers de threadpool.
- Segredos em controle de versão - Credenciais vazam via histórico do git. Correção: Carregue segredos de variáveis de ambiente ou de um cofre em tempo de execução.
- Falta de observabilidade - Incidentes são difíceis de depurar. Correção: Adicione logs estruturados, métricas e IDs de requisição.
Alternativas
| Alternativa | Use Quando | Não Use Quando |
|---|---|---|
| Framework alternativo neste cookbook | Padrão da equipe ou monólito existente | API Greenfield com restrições diferentes |
| BaaS Gerenciado | Apenas CRUD MVP | Autenticação personalizada, fluxos de trabalho ou necessidades de conformidade |
| gRPC | RPC interno de alto desempenho | Clientes HTTP públicos e acesso ao navegador |
FAQs
Quando devo adotar acesso a banco de dados assíncrono com FastAPI?
Use quando os padrões e as trocas nesta página corresponderem à sua API ou limite de dados.
Qual é o principal erro de produção com acesso a banco de dados assíncrono com FastAPI?
Ignorar validação, timeouts ou contratos de erro explícitos na borda HTTP.
Como testar o acesso a banco de dados assíncrono com FastAPI?
Use o cliente de teste do framework, substitua dependências e afirme o status mais a forma JSON.
O acesso a banco de dados assíncrono com FastAPI funciona com Python 3.14?
Sim - exemplos visam Python 3.14 com versões de framework fixadas do rodapé da pilha.
Como o acesso a banco de dados assíncrono com FastAPI se relaciona com Pydantic 2?
Valide e serialize nas bordas; mantenha os serviços funcionando com objetos de domínio tipados.
Síncrono ou assíncrono?
Prefira rotas assíncronas quando I/O domina; mantenha o trabalho de CPU pequeno ou descarregue para workers.
Onde a lógica de negócios deve residir?
Handlers finos; serviços possuem regras; repositórios possuem consultas.
Como documentar APIs?
Publique documentação OpenAPI ou de esquema que corresponda aos modelos de resposta no código.
Como lidar com versionamento?
Versionamento explícito de URL ou cabeçalho com janelas de depreciação - evite quebras silenciosas.
O que devo ler a seguir?
Siga os links Relacionados para a próxima camada de profundidade nesta seção.
Como me manter seguro?
Autentique chamadores, autorize por recurso, limite a taxa e nunca registre segredos.
Primeiro passo de performance?
Meça a latência do DB e upstream antes de trocar de framework.
Relacionado
- FastAPI Basics - Rotas e modelos principais
- Dependency Injection - Provedores por requisição
- Testing FastAPI - Padrões de cliente de teste
- Deploying FastAPI - Servindo em produção
Versões da pilha: Esta página foi escrita para Python 3.14.0 (3.14 estável, 3.13 de manutenção), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+ e uv 0.6+.