SQLAlchemy Async Engine and Async Session Explained

python sqlalchemy async engine and async session requires a clear understanding of the core syntax, runtime behavior, and practical implementation patterns demonstrated in the examples below.

SQLAlchemy's async support does not replace the core Session API. The AsyncEngine and AsyncSession classes wrap the synchronous Engine and Session, and the actual database I/O is executed on a greenlet-based bridge. You write await on the outside, but the ORM machinery inside still runs synchronously. That boundary explains most of the behavior you will observe: the connection URL must use an async driver, the session methods must be awaited, and you cannot pass an AsyncSession to code that expects a sync Session.

What the Async Engine and Async Session Wrap

create_async_engine() returns an AsyncEngine that wraps a regular Engine. Similarly, AsyncSession wraps the sync Session and exposes an almost identical API surface, but with coroutine methods. The underlying sync session is reachable through session.sync_session, which is useful when you need to run a sync-only operation inside the async session.

The bridge works through the greenlet library. When you call await session.execute(...), SQLAlchemy runs the sync DBAPI call inside a greenlet so the event loop is not blocked while the database driver waits for a response. This is why the async engine still requires an async DBAPI driver such as asyncpg or aiosqlite; the greenlet bridge cannot turn a blocking driver into a non-blocking one.

Creating the Async Engine

Use create_async_engine() with a URL that includes an async driver:

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost/appdb",
    echo=True,
)

The dialect portion of the URL must match an async driver. The following mapping is the one you will use most often:

If you pass a sync driver such as psycopg2 to create_async_engine(), SQLAlchemy raises an error because the driver does not expose an async interface. The same rule applies in reverse: create_engine("postgresql+asyncpg://...") fails because asyncpg cannot be used with the sync engine.

Building an Async Session Factory

Create sessions with async_sessionmaker(), the async counterpart of sessionmaker():

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

engine = create_async_engine("postgresql+asyncpg://user:password@localhost/appdb")
SessionFactory = async_sessionmaker(engine, expire_on_commit=False)

Setting expire_on_commit=False matters in async code. With the default True, committed objects are expired, and the next attribute access triggers a lazy refresh. That refresh is a sync operation that cannot run inside the async session context, so it raises an error. Disabling expiration avoids the trap when you need to read object attributes after commit.

The factory is cheap to create once at application startup. Each call to SessionFactory() produces a new AsyncSession bound to the same engine.

Executing Queries With an Async Session

Query execution follows the sync API but requires await on every blocking call:

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

async def get_user_by_email(session: AsyncSession, email: str):
    result = await session.execute(select(User).).where(User.email == email))
    return result.scalar_one_or_none()

For ORM rows, scalars() returns the entity objects directly:

users = (await session.execute(select(User))).scalars().all()

scalar_one_or_none() raises if the query returns more than one row, which is the right choice for a unique lookup. Use scalars().first() when you only need the first row of a potentially larger result set.

Streaming results with await session.stream() returns an AsyncResult that you iterate with async for. This avoids loading the entire result set into memory, which matters for large queries.

Transaction Boundaries and Context Managers

The cleanest transaction pattern is the nested context manager:

async with SessionFactory() as session:
    async with session.begin():
        session.add(new_user)
        session.add(new_order)

The session.begin() block commits automatically when the block exits without an exception and rolls back when an exception propagates. The outer async with guarantees session.close() runs even on failure.

When you need explicit control, use the manual commit/rollback flow:

session = SessionFactory()
try:
    await session.execute(select(User).where(User.id == user_id))
    await session.commit()
except Exception:
    await session.rollback()
    raise
finally:
    await session.close()

Never call session.commit() inside a session.begin() block; the context manager already owns the commit decision, and a second commit can lead to confusing state.

Concurrency, Connection Pooling, and Runtime Behavior

The async engine uses the same pooling machinery as the sync engine. For most dialects, the default pool is AsyncAdaptedQueuePool, which maintains a fixed set of connections and queues waiters. SQLite with aiosqlite uses NullPool by default, meaning every connection is closed when released. That is a deliberate choice because SQLite does not benefit from shared connections across threads or tasks.

You can tune the pool when you create the engine:

engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost/appdb",
    pool_size=10,
    max_overflow=20,
)

pool_size sets the number of persistent connections, and max_overflow allows temporary connections beyond that limit under load. The greenlet bridge does not block the event loop during I/O waits, but CPU-bound work inside a greenlet does block the loop. Keep heavy computation out of session callbacks and run it in a thread executor instead.

When you need to run a sync-only function, use run_sync:

def _sync_operation(sync_session):
    return sync_session.execute(text("SELECT 1"))

result = await session.run_sync(_sync_operation)

This is the supported escape hatch for operations that have no async equivalent, such as certain reflection or dialect-specific calls.

Common Failure Modes When Mixing Sync and Async

Several errors come up repeatedly when developers first move to the async API.

Missing await. session.execute(...) returns a coroutine. If you forget await, the query never runs and you get a RuntimeWarning about an unawaited coroutine. The same applies to commit(), rollback(), and close().

Sync Session with an async engine. Constructing Session(engine) where engine is an AsyncEngine raises an error because AsyncEngine is not a valid sync Engine. Use AsyncSession or the factory instead.

Async driver with a sync engine. create_engine("postgresql+asyncpg://...") fails because asyncpg only exposes an async interface. Match the driver to the engine type.

Lazy loading after commit. With expire_on_commit=True, accessing user.orders after a commit triggers a lazy load that runs synchronously and raises MissingGreenlet. Set expire_on_commit=False and eagerly load relationships you need after commit, or access them before the transaction ends.

Production Considerations for Async SQLAlchemy

Shut down the engine cleanly when your application stops. The async engine holds a connection pool, and an abrupt exit can leave sockets open:

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app):
    yield
    await engine.dispose()

await engine.dispose() closes all pooled connections and waits for in-flight operations to finish. In a FastAPI application, attach this to the lifespan handler so the pool is released on shutdown.

For observability, set echo=True during development to log SQL statements. In production, attach event listeners to the engine's before_cursor_execute and after_cursor_execute events to measure query latency without logging every statement. The pool itself can be monitored through engine.pool.status(), which reports checked-out and checked-in connection counts, useful for detecting connection leaks.

Finally, keep the session scope narrow. Create a session per request or per unit of work, run the transaction, and close it. Long-lived AsyncSession instances that span multiple requests tend to accumulate expired state and make lazy-loading errors harder to diagnose.