Skip to content

Scoping Database Sessions with contextvars

Passing a database session through every function signature is tedious, so most async applications put "the current session" somewhere ambient. Thread-locals do not work in asyncio — every request on the loop shares one thread — and context variables are the correct replacement: each request task gets its own binding, and code anywhere below the handler can find the session. The pattern has one trap specific to async code. Child tasks inherit the context, so a TaskGroup fan-out inside a request hands the same session to every child, and an AsyncSession does not allow concurrent use: with SQLAlchemy 2.1 and five children issuing select 1, the group failed with InvalidRequestError: This session is provisioning a new connection; concurrent operations are not permitted. Giving each child its own session through the same context mechanism fixed it — all five returned 1 and the outer session kept working.

Prerequisites

1. Put the session in a ContextVar with a scope

One context variable, one accessor, and one context manager that opens, binds, commits or rolls back, and unbinds:

import contextvars
from contextlib import asynccontextmanager

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

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

_session: contextvars.ContextVar[AsyncSession | None] = contextvars.ContextVar("db_session", default=None)


def db() -> AsyncSession:
    s = _session.get()
    if s is None:
        raise RuntimeError("no database session in this context")
    return s


@asynccontextmanager
async def session_scope():
    async with Session() as s:
        token = _session.set(s)
        try:
            async with s.begin():          # commit on success, roll back on exception
                yield s
        finally:
            _session.reset(token)

The default is None, not a session, so code that runs outside any scope fails loudly — verified: calling db() after the scope ended raised no database session in this context. A default of an actual session object would be shared by every request in the process.

Verify: inside session_scope(), db() returns the session; outside, it raises.

Where the session lives during a request 4 stacked layers. Where the session lives during a request middleware or dependency session_scope(): open, begin, bind request task context _session -> this request's AsyncSession repositories and services db() — no session parameter engine pool one connection checked out per active session The context variable replaces a parameter, not the session's lifecycle; the scope still owns open, commit and close.

2. Open the scope at the edge of each request

The scope belongs where the request starts. In FastAPI, a dependency with yield does it and composes with routing:

from fastapi import Depends, FastAPI

app = FastAPI()


async def request_session():
    async with session_scope() as s:
        yield s


@app.post("/orders", dependencies=[Depends(request_session)])
async def create_order(payload: OrderIn):
    order = await orders_repo.create(payload)     # uses db() internally
    await audit_repo.record("order.created", order.id)
    return order


# repositories never take a session argument
class OrdersRepo:
    async def create(self, payload) -> Order:
        order = Order(**payload.model_dump())
        db().add(order)
        await db().flush()
        return order

Two requests running concurrently on the loop are two tasks, so each has its own binding; nothing can cross. One caveat for frameworks: the dependency must run in the same task as the handler for the binding to be visible, which FastAPI's yield dependencies do. Starlette BackgroundTasks run after the response in a context captured earlier — by then the scope has closed, and db() fails, which is the correct outcome. How yield dependencies interact with the response lifecycle is covered in managing async resources with FastAPI yield dependencies.

Verify: two concurrent requests that each insert a row commit independently, and an exception in one rolls back only that one.

3. Do not share one session across child tasks

Fan-out inside a request is where the pattern breaks. TaskGroup children copy the parent's context, so they all see the same session:

async def dashboard():
    async with asyncio.TaskGroup() as tg:
        a = tg.create_task(orders_repo.recent())        # all three use db() ...
        b = tg.create_task(stats_repo.totals())          # ... which is the SAME session
        c = tg.create_task(users_repo.active())

An AsyncSession wraps one connection and one transaction; it is not designed for concurrent operations, and SQLAlchemy refuses: measured with five children, InvalidRequestError: This session is provisioning a new connection; concurrent operations are not permitted. Even where a driver does not detect it, interleaving statements on one connection produces wrong results or protocol errors. The explanation from the driver side is in why one SQLAlchemy AsyncSession cannot run queries concurrently.

Verify: reproduce the error once with your driver so you recognise it.

Three children, one session, one connection A sequence of 4 messages between 4 participants. Three children, one session, one connection child A child B shared AsyncSession connection execute(recent orders) checking out connection execute(totals) at the same time InvalidRequestError: concurrent operations Context inheritance hands every child the parent's session; the session is not built for concurrent use.

4. Give each child its own scope

Wrap each child in its own session_scope(). Because the scope sets the context variable inside the child task, the child's db() resolves to its own session, and the parent's binding is untouched:

async def in_own_session(fn, *args):
    async with session_scope():
        return await fn(*args)


async def dashboard():
    async with asyncio.TaskGroup() as tg:
        a = tg.create_task(in_own_session(orders_repo.recent))
        b = tg.create_task(in_own_session(stats_repo.totals))
        c = tg.create_task(in_own_session(users_repo.active))
    return a.result(), b.result(), c.result()

Verified with five children: each returned its result, and the parent's db() still worked afterwards. The cost is one pooled connection per child for the duration of the fan-out, so size the pool for peak fan-out, not peak requests — a request that fans out to five queries holds up to six connections at once. The arithmetic is in sizing async connection pools for throughput.

Each child also gets its own transaction. If the children must see one consistent snapshot, either run them sequentially on the request session or use a REPEATABLE READ snapshot that can be shared — concurrent and transactionally identical are not available together on one connection.

Verify: under load, the pool's checked-out count peaks at about requests × (1 + fan-out width).

Queries inside one request: shared or separate sessions? A decision on Must the queries share one transaction with 3 outcomes. Queries inside one request: shared or separate sessions? Must the queries share one transaction? yes sequential awaits on the request session no, independent reads own session_scope per child pool sized for fan-out concurrent on one session never InvalidRequestError Concurrency and one shared transaction are mutually exclusive on a single connection.

5. Keep background work out of the request's session

A task spawned from a request inherits the request's context — and therefore its session — but outlives the request. When the request's scope closes, the background task holds a closed session (and, if it runs before the close, a session it shares concurrently with the handler):

async def handle_upload(req):
    file = await files_repo.save(req.file)
    spawn_detached(index_file(file.id), name=f"index:{file.id}")   # starts with an empty context
    return {"id": file.id}


async def index_file(file_id: int):
    async with session_scope():                     # its own session, its own transaction
        file = await files_repo.get(file_id)
        await search.index(file)

spawn_detached creates the task with context=contextvars.Context(), as described in running callbacks in a copied context, so the background job cannot reach the request's session by accident — db() raises until the job opens its own scope.

Verify: a background job that forgets to open a scope fails immediately with "no database session in this context" instead of using a stale session.

Verification

Session scoping is correct when:

  • Every request runs in exactly one session_scope(), opened at the edge.
  • db() outside a scope raises, and nothing has a session as the ContextVar default.
  • Concurrent children each open their own scope, and the pool is sized for the fan-out.
  • Background tasks start with an empty context and open their own scope.

Diagnostic Hook: export the engine pool's checked-out connections and overflow as gauges, and count InvalidRequestError with "concurrent operations" in the message. That error rate should be zero; any occurrence is a child task using an inherited session. A checked-out gauge that sits near the pool size during fan-out-heavy endpoints means the pool is sized for requests but not for fan-out.

Pitfalls & edge cases

  • Session as the ContextVar default. Shared by every request in the process.
  • async_scoped_session keyed on the current task. It works but ties sessions to task identity, so every child task silently gets a new session; explicit scopes make the boundary visible.
  • Forgetting expire_on_commit=False. Accessing attributes after commit triggers lazy loads, which fail in async code with a MissingGreenlet error.
  • Spawning background work from inside a scope without an empty context — it inherits a session that is about to close.

Frequently Asked Questions

How do I make the current database session available without passing it everywhere?

Store the request's AsyncSession in a ContextVar inside a context manager that opens the session, sets the variable, and resets it with the token on exit. Code below the request handler calls an accessor that reads the variable.

Why do I get 'concurrent operations are not permitted' with AsyncSession?

Several tasks are using the same session at once — usually TaskGroup or gather children that inherited it through the context. An AsyncSession wraps one connection; give each concurrent child its own session.

Can I use thread-local sessions with asyncio?

No. All tasks on an event loop share one thread, so a thread-local session would be shared by every concurrent request. Context variables give each task its own binding.

Do background tasks inherit the request's database session?

Yes, if they are created inside the request's context, which is a bug once the request's session closes. Create them with an empty context and open a new session scope inside the background task.