Native Async Support

SQLAlchemy 2.0 stabilizes async support via AsyncSession and create_async_engine. Combined with asyncpg or aiosqlite, it lets you use SQLAlchemy in FastAPI or asyncio applications without blocking the event loop.

Example with asyncpg

python
from sqlalchemy.ext.asyncio import (
    AsyncSession, create_async_engine, async_sessionmaker
)
from sqlalchemy import select

engine = create_async_engine(
    'postgresql+asyncpg://user:pass@localhost/db'
)
async_session = async_sessionmaker(engine, class_=AsyncSession)

async def get_users():
    async with async_session() as session:
        stmt = select(User).where(User.active == True)
        result = await session.execute(stmt)
        return result.scalars().all()

# Usable in FastAPI:
# @app.get('/users')
# async def users():
#     return await get_users()

Sources