Bancos de Dados Assíncronos
O acesso assíncrono ao banco de dados mantém o loop de eventos livre enquanto espera pelo servidor. asyncpg (PostgreSQL), SQLAlchemy 2 async e pools específicos do driver integram-se com o lifespan do FastAPI e sessões por requisição.
Receita
# SQLAlchemy 2 async + asyncpg Conceitual
# engine = create_async_engine("postgresql+asyncpg://...")
# async with AsyncSession(engine) as session:
# result = await session.execute(select(User).limit(1))Quando usar isso:
- Rotas do FastAPI acessando PostgreSQL sob concorrência
- Aplicações de long-poll ou websocket compartilhando pool de banco de dados
- Microsserviços com cargas de trabalho de consulta limitadas por I/O
- Substituindo SQLAlchemy síncrono em rotas
async def - Ajuste do pool de conexões para limites de contêineres
Exemplo Funcional
Fluxo simulado no estilo asyncpg com substitutos do asyncio para demonstração executável sem um banco de dados real.
import asyncio
from dataclasses import dataclass, field
@dataclass
class FakePool:
_connections: int = 0
max_size: int = 5
async def acquire(self) -> "FakeConn":
await asyncio.sleep(0.01)
self._connections += 1
return FakeConn(self)
async def release(self, conn: "FakeConn") -> None:
await asyncio.sleep(0)
self._connections -= 1
@dataclass
class FakeConn:
pool: FakePool
async def fetchval(self, query: str) -> int:
await asyncio.sleep(0.02)
return 42
class DB:
def __init__(self, pool: FakePool) -> None:
self._pool = pool
async def user_count(self) -> int:
conn = await self._pool.acquire()
try:
return await conn.fetchval("SELECT COUNT(*) FROM users")
finally:
await self._pool.release(conn)
async def main() -> None:
pool = FakePool()
db = DB(pool)
counts = await asyncio.gather(*(db.user_count() for _ in range(3)))
print(counts, "pool conns", pool._connections)
asyncio.run(main())O que isso demonstra:
- O padrão acquire/release espelha pools reais
finallygarante que a conexão retorne ao pool- Consultas concorrentes sobrepõem esperas em um único loop
- Rastreia o tamanho do pool para evitar esgotar as conexões máximas do banco de dados
Mergulho Profundo
Opções de Stack
| Camada | PostgreSQL |
|---|---|
| Driver | asyncpg |
| ORM | SQLAlchemy 2 AsyncSession |
| Migrações | Alembic (engine síncrono) trabalho separado |
Escopo da Sessão
- Prefira uma sessão por requisição
- Commit no caminho de requisição bem-sucedido; rollback em caso de exceção
- Não compartilhe a sessão entre tarefas concorrentes
Armadilhas
- Sessão síncrona em rota assíncrona - bloqueia o loop. Correção:
AsyncSession+await execute. - Pool por requisição - tempestade de conexões. Correção: engine/pool escopado na aplicação no lifespan.
awaitfaltando na execução - retorna corrotina, bugs lógicos silenciosos. Correção: regras ASYNC do pyright/ruff.- Consultas N+1 em async - ainda ruins com paralelismo. Correção: carregamento antecipado (eager loading), consultas em lote.
- Transações abrangendo awaits para APIs externas - locks longos. Correção: transações curtas de banco de dados apenas em torno do trabalho do banco de dados.
Alternativas
| Alternativa | Usar Quando | Não Usar Quando |
|---|---|---|
| ORM Síncrono + thread pool | Migração gradual | API assíncrona Greenfield |
| asyncpg puro | Desempenho máximo, controle SQL | Necessidades de ORM pesado |
| SDK async do DynamoDB | AWS NoSQL | Necessidades relacionais |
FAQs
asyncpg vs psycopg3 async?
Ambos são viáveis; SQLAlchemy abstrai o driver - escolha com base na familiaridade operacional e necessidades de recursos.
Fórmula do tamanho do pool?
(requisições_concorrentes_esperadas * tempo_medio_consulta) / latencia_alvo limitado por max_connections do banco de dados.
Padrão SQLAlchemy 2?
create_async_engine, async_sessionmaker, async with session.begin().
Migrações async?
Alembic geralmente executa um trabalho de migração síncrono em CI - não no caminho crítico de requisição.
SQLite async?
aiosqlite para desenvolvimento/aplicativos pequenos - limites de concorrência diferentes do Postgres.
ORM async do Django?
Melhorando - verifique a documentação do Django 5.2 para suporte a consultas async em seus modelos.
Como testar?
Testcontainers Postgres ou sqlite+aiosqlite com fixtures pytest-asyncio.
Vazamentos de conexão?
Sempre gerencie sessões com gerenciadores de contexto; monitore a métrica de contagem de checked-out do pool.
Réplicas de leitura?
Engines/pools separados para roteamento de leitura vs. escrita na camada de aplicação.
Pydantic com linhas?
model_validate em linhas de dicionário de mapping().all() - valide na fronteira.
Relacionado
- Gerenciadores e Iteradores de Contexto Assíncronos - escopos de sessão
- Concorrência Estruturada - escopo de tarefa de requisição
- Ponte Síncrona/Assíncrona - evite ORM síncrono
- Rotas assíncronas do FastAPI - fiação do framework
Versões da Stack: Esta página foi escrita para Python 3.14.0 (estável 3.14, manutenção 3.13), 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+.