TopolearnAsync web frameworks like FastAPI need non-blocking database access, and async has rules the sync ORM doesn't.
The sqlalchemy.ext.asyncio extension wraps the same Core and ORM in an async API. Install it with pip install "sqlalchemy[asyncio]", which adds the greenlet dependency, plus an async DBAPI driver: postgresql+asyncpg://, mysql+aiomysql:// or sqlite+aiosqlite://.
engine = create_async_engine(url);async with engine.begin() as conn:for Core work, andawait conn.run_sync(Base.metadata.create_all)to call sync-only helpers.async_sessionmaker(engine, expire_on_commit=False)createsAsyncSessions. Disabling expire-on-commit matters more here, because reloading an expired attribute would require IO.- Every operation that talks to the database is awaited:
await session.execute(stmt),await session.scalars(stmt),await session.get(User, 1),await session.commit().session.add()is not awaited; it only stages.
The big rule: no implicit IO. Accessing a lazy-loaded relationship can't run a query from a plain attribute access, so it raises MissingGreenlet. Load what you need up front with selectinload, or use lazy="raise" to find mistakes; the AsyncAttrs mixin also offers await obj.awaitable_attrs.posts for an explicit async load.
One AsyncSession per task: it's not safe to use concurrently from multiple coroutines, so asyncio.gather over one session is a bug. Call await engine.dispose() at shutdown.
Example
import asyncio
from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
class Base(DeclarativeBase): pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str]
async def main():
engine = create_async_engine("sqlite+aiosqlite://")
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
Session = async_sessionmaker(engine, expire_on_commit=False)
async with Session() as s:
s.add_all([User(name="ada"), User(name="bob")]) # no await: just staging
await s.commit()
names = (await s.scalars(select(User.name).order_by(User.name))).all()
print(names) # ['ada', 'bob']
await engine.dispose()
asyncio.run(main())Remember
- Async engine + async driver (asyncpg, aiosqlite) + AsyncSession
- Await execute/scalars/get/commit; add() is synchronous
- No lazy loading: eager-load relationships or use awaitable_attrs
- One AsyncSession per task; never share it across concurrent coroutines
Watch out for
- Touching user.posts in async code and getting MissingGreenlet.
- Running asyncio.gather on several queries using the same AsyncSession.