SQLAlchemy Async en Python: Engine y AsyncSession

El soporte asyncio de SQLAlchemy adapta Core y ORM a drivers asíncronos. La sesión sigue siendo stateful: cada tarea concurrente necesita su propia AsyncSession.

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

engine = create_async_engine("postgresql+asyncpg://user:pass@db/app")
Session = async_sessionmaker(engine, expire_on_commit=False)


async def buscar_usuario(user_id: int):
    async with Session() as session:
        async with session.begin():
            return await session.scalar(
                select(Usuario).where(Usuario.id == user_id)
            )

No versiones credenciales; carga la URL mediante configuración segura. session.begin() delimita commit y rollback. Cierra el engine al terminar con await engine.dispose().

Evita lazy loading que dispara I/O implícito fuera de un await. Planifica relaciones y no llames código bloqueante en el loop. Consulta SQLAlchemy y Alembic.

La documentación oficial de asyncio en SQLAlchemy, consultada el 22 de julio de 2026, cubre drivers, sesiones y concurrencia. Adopta async cuando todo el flujo pueda ser no bloqueante.

Una sesión por solicitud o tarea

Crea el engine una vez al iniciar y comparte el async_sessionmaker. Abre una AsyncSession nueva para cada solicitud, job en segundo plano o tarea concurrente. Cerrar la sesión al terminar la unidad de trabajo forma parte del contrato: la conexión vuelve al pool y se limpia el identity map.

No guardes una sesión en un servicio de larga duración y la reutilices entre awaits de tareas distintas. El uso concurrente de una sola sesión puede corromper la transacción o generar errores difíciles de diagnosticar. Pasa la sesión como dependencia, o ábrela en el borde del caso de uso e inyecta repositorios que la reciben.

expire_on_commit=False es habitual en apps web para que los objetos sigan usables tras el commit sin un refresh inmediato. Si necesitas el estado fresco de la base, llama await session.refresh(obj) o ejecuta una nueva consulta. Prefiere refresh explícito a lazy loads accidentales.

Transacciones y límites de commit

async with session.begin() inicia la transacción y hace commit si todo va bien o rollback si hay excepción. El trabajo anidado debe usar la misma sesión y la misma transacción, salvo que necesites otra conexión a propósito. Evita mezclar session.commit() dispersos en el dominio con context managers externos que también hacen commit.

Cuando un caso de uso abarca varias escrituras que deben tener éxito juntas, mantenlas en el mismo bloque begin(). Si un paso posterior falla, los inserts anteriores deben revertirse. Para consultas de solo lectura, basta una sesión corta sin transacción de escritura, pero cierra la sesión con rapidez.

Trata los errores de integridad en la frontera y asígnalos a resultados de dominio como conflicto o validación. No expongas texto específico del driver a los clientes de la API.

Carga relaciones sin I/O implícito

SQLAlchemy async desaconseja el lazy loading porque el acceso a atributos no puede ejecutar I/O con seguridad. Carga lo necesario con selectinload, joinedload o consultas explícitas antes de salir de la sesión:

from sqlalchemy.orm import selectinload


async def buscar_pedido(session, pedido_id: int):
    stmt = (
        select(Pedido)
        .where(Pedido.id == pedido_id)
        .options(selectinload(Pedido.items))
    )
    return await session.scalar(stmt)

Decide la estrategia de carga por caso de uso. Traer grafos demasiado grandes desperdicia ancho de banda; traer de menos obliga a consultas extra o a un acceso roto cuando la sesión ya está cerrada. Convierte objetos ORM en schemas o DTOs antes de devolver la respuesta cuando esta viva más que la sesión.

Engine, pool y elección de driver

create_async_engine necesita un driver compatible con el dialecto, como asyncpg para PostgreSQL o aiosqlite para pruebas locales con SQLite. El tamaño del pool, el overflow y el recycle siguen importando bajo carga async: demasiados checkouts concurrentes esperan o fallan, mientras que un pool excesivo puede saturar la base.

Configura timeouts de statement y conexión según el driver y el despliegue. Una consulta colgada sin timeout puede agotar el pool aunque el event loop siga respondiendo. Mide por separado la espera de checkout y la duración de la consulta.

No envuelvas un Session síncrono o una llamada bloqueante de DB-API en async def y lo llames desde el event loop. Si parte de la pila sigue siendo síncrona, ejecútala en un worker thread o mantén ese camino totalmente sync.

Probar repositorios async

Usa el mismo driver async que te importa para las pruebas de integración, o un contenedor con una base descartable. Un repositorio fake en memoria sirve para reglas de dominio, pero no demuestra SQL, constraints ni comportamiento de transacción.

import pytest


@pytest.fixture
async def session(engine):
    async with engine.connect() as connection:
        transaction = await connection.begin()
        Session = async_sessionmaker(bind=connection, expire_on_commit=False)
        async with Session() as session:
            yield session
        await transaction.rollback()

Confirma que los límites de conexión y transacción de la fixture coinciden con la forma en que la aplicación abre sesiones. El código que crea un segundo engine o hace commit en otra conexión puede escapar del rollback externo. Prefiere schemas únicos o bases recreadas cuando el aislamiento deba ser más fuerte.

Lista operativa

Haz dispose del engine al apagar. Mantén las credenciales fuera del código y de los logs. Prefiere carga explícita al acceso lazy. Limita la concurrencia para que las tareas no retiren más conexiones de las que el pool puede servir. Cubre éxito, conflictos de integridad, resultados vacíos y limpieza tras una cancelación.

SQLAlchemy async ayuda cuando el resto del camino de la solicitud ya es asíncrono. La ganancia es I/O cooperativo, no velocidad mágica de consulta. La propiedad clara de la sesión y las transacciones explícitas importan más que convertir cada def en async def.