Bases de datos asíncronas
El acceso asíncrono a la base de datos mantiene el bucle de eventos libre mientras se espera la respuesta del servidor. asyncpg (PostgreSQL), SQLAlchemy 2 async y los pools específicos del driver se integran con el ciclo de vida de FastAPI y las sesiones por solicitud.
Receta
# SQLAlchemy 2 async + asyncpg conceptual
# engine = create_async_engine("postgresql+asyncpg://...")
# async with AsyncSession(engine) as session:
# result = await session.execute(select(User).limit(1))Cuándo usar esto:
- Rutas de FastAPI que acceden a PostgreSQL bajo concurrencia
- Aplicaciones de long-poll o websocket que comparten un pool de bases de datos
- Microservicios con cargas de trabajo de consulta intensivas en E/S
- Reemplazar SQLAlchemy síncrono en rutas
async def - Ajuste del pool de conexiones para límites de contenedores
Ejemplo funcional
Flujo simulado estilo asyncpg con sustitutos de asyncio para una demostración ejecutable sin una base de datos 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())Lo que esto demuestra:
- El patrón de adquisición/liberación imita a los pools reales
finallyasegura que la conexión regrese al pool- Las consultas concurrentes solapan las esperas en un solo bucle
- Rastrea el tamaño del pool para evitar agotar las conexiones máximas de la base de datos
Profundización
Opciones de Stack
| Capa | PostgreSQL |
|---|---|
| Driver | asyncpg |
| ORM | SQLAlchemy 2 AsyncSession |
| Migraciones | Alembic (motor síncrono) trabajo separado |
Ámbito de Sesión
- Prefiere una sesión por solicitud
- Confirma en la ruta de solicitud exitosa; revierte en caso de excepción
- No compartas la sesión entre tareas concurrentes
Errores comunes
- Sesión síncrona en ruta asíncrona: bloquea el bucle. Solución:
AsyncSession+await execute. - Pool por solicitud: tormenta de conexiones. Solución: motor/pool a nivel de aplicación en el ciclo de vida.
- Falta
awaitenexecute: devuelve una corrutina, errores lógicos silenciosos. Solución: reglas ASYNC de pyright/ruff. - Consultas N+1 en async: siguen siendo malas con paralelismo. Solución: carga anticipada, consultas por lotes.
- Transacciones que abarcan
awaita APIs externas: bloqueos largos. Solución: transacciones de base de datos cortas solo alrededor del trabajo de la base de datos.
Alternativas
| Alternativa | Usar cuando | No usar cuando |
|---|---|---|
| ORM síncrono + pool de hilos | Migración gradual | API asíncrona nueva |
| asyncpg puro | Máximo rendimiento, control SQL | Necesidades de ORM complejas |
| SDK asíncrono de DynamoDB | AWS NoSQL | Necesidades relacionales |
Preguntas frecuentes
¿asyncpg vs psycopg3 async?
Ambos son viables; SQLAlchemy abstrae el driver - elige según la familiaridad operativa y las necesidades de características.
¿Fórmula del tamaño del pool?
(solicitudes_concurrentes_esperadas * tiempo_promedio_consulta) / latencia_objetivo limitado por las conexiones_max de la base de datos.
¿Patrón SQLAlchemy 2?
create_async_engine, async_sessionmaker, async with session.begin().
¿Migraciones asíncronas?
Alembic típicamente ejecuta un trabajo de migración síncrono en CI - no en la ruta de acceso activa de la solicitud.
¿SQLite async?
aiosqlite para desarrollo/aplicaciones pequeñas - límites de concurrencia diferentes a los de Postgres.
¿ORM asíncrono de Django?
Mejorando - consulta la documentación de Django 5.2 para soporte de consultas asíncronas en tus modelos.
¿Cómo probar?
Testcontainers Postgres o sqlite+aiosqlite con fixtures de pytest-asyncio.
¿Fugas de conexión?
Siempre gestiona las sesiones con gestores de contexto; monitoriza la métrica de recuento de conexiones en uso del pool.
¿Réplicas de lectura?
Motores/pools separados para enrutamiento de lectura vs escritura en la capa de aplicación.
¿Pydantic con filas?
model_validate en diccionarios de filas de mapping().all() - valida en el límite.
Relacionado
- Gestores y Iteradores de Contexto Asíncronos - ámbitos de sesión
- Concurrencia Estructurada - ámbito de tarea de solicitud
- Puente Síncrono/Asíncrono - evitar ORM síncrono
- Rutas asíncronas de FastAPI - cableado del framework
Versiones del Stack: Esta página fue escrita para Python 3.14.0 (estable 3.14, mantenimiento 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+, y uv 0.6+.