Using async_scoped_session with asyncio tasks

async_scoped_session(factory, scopefunc=asyncio.current_task) gives one session per task rather than per thread, which is the only correct scope under asyncio — but in a framework that already scopes a session per request, a plain async_sessionmaker passed explicitly is simpler and harder to leak. This is the scoping question under session lifecycle and scope management.

Quick Answer

Before — a thread-local scope under asyncio:

from sqlalchemy.ext.asyncio import async_scoped_session, async_sessionmaker

Session = async_scoped_session(async_sessionmaker(engine))   # thread-scoped: wrong

After — scoped on the current task:

from __future__ import annotations

import asyncio

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

engine = create_async_engine("postgresql+asyncpg://app:secret@db.internal/orders")
factory = async_sessionmaker(engine, expire_on_commit=False)

# One session per asyncio task, not per thread.
Session = async_scoped_session(factory, scopefunc=asyncio.current_task)


async def handle_job(job_id: int) -> None:
    try:
        session: AsyncSession = Session()          # same object anywhere in this task
        async with session.begin():
            await process(session, job_id)
    finally:
        await Session.remove()                     # closes it and clears the registry

The finally is not optional. The registry holds a strong reference keyed on the task, so a missing remove() leaks the session, its identity map and — until it is closed — its pooled connection.

Execution Context & Async Workflow Integration

asyncio.current_task is the right scopefunc because it identifies the unit that asyncio actually schedules. It has one consequence worth understanding: a child task is a different task, so it gets a different session.

# Async — a child task does not inherit the parent's session
async def parent() -> None:
    session = Session()                    # session A
    async with asyncio.TaskGroup() as group:
        group.create_task(child())         # gets session B, not A
    await Session.remove()                 # removes A only


async def child() -> None:
    try:
        session = Session()                # session B, its own connection
        ...
    finally:
        await Session.remove()             # B must remove itself

That is correct — two concurrent tasks must not share an AsyncSession, as [session lifecycle and scope management](' + C + ') sets out — but it surprises people who expect scoping to be inherited. Each task owns its session and each task must remove it, which means every spawned task needs its own try/finally.

It also means a scoped registry does not save you from thinking about concurrency; it saves you from passing an argument. If the code being called does not spawn tasks, that is a genuine simplification. If it does, the registry adds a cleanup obligation at every spawn point, and an explicit argument is usually clearer.

In a framework that already provides request scoping, the registry is redundant:

# FastAPI already gives one session per request, with cleanup — no registry needed
async def get_session() -> AsyncIterator[AsyncSession]:
    async with factory() as session:
        yield session


@app.get("/orders")
async def list_orders(session: AsyncSession = Depends(get_session)):
    ...

Resolving Warnings, Errors & Common Mistakes

Advanced: Choosing a scopefunc, and When Not To

asyncio.current_task is the right default, and two variants come up.

A request-scoped ContextVar keys the session on a request identifier rather than a task, so background work spawned by a request shares the request's session. That is occasionally what you want and is usually a trap: the background task then outlives the request, and the session it is sharing has been closed.

import contextvars

request_id: contextvars.ContextVar[str] = contextvars.ContextVar("request_id")
Session = async_scoped_session(factory, scopefunc=request_id.get)

A worker-scoped function in a batch process, where each worker coroutine handles many items and one session per worker is the desired granularity — which makes the unit of work the worker rather than the item, and is a deliberate transactional decision rather than a scoping convenience.

The more useful question is whether to use a registry at all. It earns its place when a session must be reachable from code that cannot take one as an argument: a deep call stack, a logging filter, a legacy layer. It does not earn its place in a modern web application, where the framework already creates and disposes of a session per request and passing it explicitly makes every dependency visible.

The cost is the remove(). Every path out of the task — return, exception, cancellation — must reach it, which means a try/finally at every entry point rather than one context manager. A single missed path is a slow leak that shows up as pool exhaustion hours later, and the pool counters in instrumenting and observing async queries are what make it visible before it becomes an outage.

If you do adopt a registry, wrap it so the cleanup cannot be forgotten:

from contextlib import asynccontextmanager


@asynccontextmanager
async def scoped_session() -> AsyncIterator[AsyncSession]:
    """The only supported way to use the registry: cleanup is structural."""
    try:
        yield Session()
    finally:
        await Session.remove()

With that in place, async with scoped_session() as session: is as safe as the explicit form and keeps the registry's reachability benefit — which is the arrangement worth aiming for if the registry is genuinely needed.

Migrating Away From a Scoped Session

A codebase that adopted scoped_session in the synchronous era usually finds that the async migration is a good moment to remove it, because the reason it existed — a thread-local session reachable from anywhere — no longer maps onto the execution model.

The migration is mechanical and can be done incrementally. Start at the outermost layer: the request handler or job entry point already knows where the unit of work begins, so give it an explicit session and pass it one level down. Repeat inward. At each step the registry is still available for the layers not yet converted, so nothing breaks in between.

# Step 1 — the entry point creates it explicitly and passes it down
async def handle_job(job_id: int) -> None:
    async with factory() as session, session.begin():
        await process(session, job_id)       # process() now takes a session


# Step 2 — process() passes it further rather than reaching for the registry
async def process(session: AsyncSession, job_id: int) -> None:
    order = await load_order(session, job_id)
    await apply_pricing(session, order)

Two signals tell you the conversion is finished. Nothing outside the composition root references the registry, which a grep confirms. And the tests get simpler: a function taking a session can be handed the rollback-isolated test session directly, where a function reaching for a registry needs that registry configured, which is exactly the friction described in testing async SQLAlchemy applications.

The one place worth keeping a registry is a framework integration you do not own — a plugin interface that calls your code with no way to pass anything. There, wrap it in the context manager above and treat it as an adapter rather than as the application's normal way of reaching a session.

Frequently Asked Questions

Related

• Session Lifecycle and Scope Management — The parent guide: what a session owns, and why one per concurrent task is the rule.
• Fixing DetachedInstanceError after commit in SQLAlchemy — What happens to instances when the scoped session is finally removed.
• Testing Async SQLAlchemy Applications — Why an explicitly passed session is dramatically easier to test.
• Instrumenting and observing async queries — The pool counters that reveal a leaked session before it exhausts the pool.