Running background tasks with a fresh AsyncSession in FastAPI

Pass primary keys to background_tasks.add_task(), never the request's AsyncSession or ORM objects, and open a new session from your async_sessionmaker inside the task — the dependency that yielded the request session closes it as the request ends. This guide belongs to integrating SQLAlchemy async with FastAPI and Starlette.

Quick Answer

A background task runs after the response. By then the request's session is closed or closing, and objects loaded through it are detached.

Before — the request session and an ORM object handed to the task:

from fastapi import BackgroundTasks, Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession

from shop.db import get_session
from shop.mail import send_email
from shop.models import Order

app = FastAPI()


async def send_receipt(session: AsyncSession, order: Order) -> None:
    # order.customer is not loaded; the session that could load it is gone.
    await send_email(order.customer.email, f"Receipt for order {order.id}")
    order.receipt_sent = True
    await session.commit()


@app.post("/orders")
async def create_order(
    payload: dict,
    background_tasks: BackgroundTasks,
    session: AsyncSession = Depends(get_session),
):
    order = Order(**payload)
    session.add(order)
    await session.commit()
    background_tasks.add_task(send_receipt, session, order)
    return {"id": order.id}
# sqlalchemy.orm.exc.DetachedInstanceError: Parent instance <Order at 0x7f...> is not bound
# to a Session; lazy load operation of attribute 'customer' cannot proceed

After — an ID in, a fresh session inside:

from fastapi import BackgroundTasks, Depends, FastAPI
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from shop.db import Session, get_session
from shop.mail import send_email
from shop.models import Order

app = FastAPI()


async def send_receipt(order_id: int) -> None:
    async with Session() as session:
        order = await session.scalar(
            select(Order).where(Order.id == order_id).options(selectinload(Order.customer))
        )
        if order is None or order.receipt_sent:
            return
        await send_email(order.customer.email, f"Receipt for order {order.id}")
        order.receipt_sent = True
        await session.commit()


@app.post("/orders")
async def create_order(
    payload: dict,
    background_tasks: BackgroundTasks,
    session: AsyncSession = Depends(get_session),
):
    order = Order(**payload)
    session.add(order)
    await session.commit()
    background_tasks.add_task(send_receipt, order.id)
    return {"id": order.id}

Session here is the application's async_sessionmaker, the same factory get_session uses. The receipt_sent check makes the task safe to run twice, which matters as soon as it is retried or moved to a queue.

Execution Context & Async Workflow Integration

FastAPI runs BackgroundTasks in the same process and on the same event loop as the request, after the response has been produced. They are a Starlette feature: the response object carries the tasks, and Starlette awaits them once the body has been sent. That makes them cheap — no queue, no worker — and it places them outside the request's scope.

The request session is scoped by a dependency with yield:

from collections.abc import AsyncIterator

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

engine = create_async_engine("postgresql+asyncpg://shop:secret@db/shop", pool_size=10)
Session = async_sessionmaker(engine, expire_on_commit=False)


async def get_session() -> AsyncIterator[AsyncSession]:
    async with Session() as session:
        yield session

When the async with block exits, the session closes and its connection returns to the pool. Exactly when that happens relative to background tasks has changed between FastAPI releases — some versions run dependency cleanup before the response is sent, others after — and code that depends on either ordering breaks on upgrade. Treat the request session as unavailable to anything scheduled with add_task(), whatever the version.

The failures that result depend on what the task does first. Touching an unloaded attribute on a passed object raises DetachedInstanceError, or under async MissingGreenlet if the object is still nominally attached. Executing on a session that the dependency is closing at the same moment raises IllegalStateChangeError: Method 'close()' can't be called here. Executing on a session that has already closed can even appear to work, because a closed AsyncSession in 2.0 starts a new transaction on next use — and then nothing closes it, and the connection it checked out is returned only when the garbage collector finds the session, with the warning covered in fixing garbage collector non-checked-in connection warnings.

A fresh session inside the task avoids all three, and it has a correctness benefit too: the task reloads committed data. If the request's transaction had rolled back, the reload finds nothing and the task exits instead of emailing a receipt for an order that does not exist. The broader rule — one session per unit of concurrent work — is the subject of running concurrent queries with AsyncSession.

Resolving Warnings, Errors & Common Mistakes

The last row is operational rather than a SQLAlchemy error, and it is the one that hides failures longest. A background task that raises after the response has no client to report to. Wrap task bodies with explicit reporting:

import functools
import logging
from collections.abc import Awaitable, Callable
from typing import ParamSpec

log = logging.getLogger("background")
P = ParamSpec("P")


def reported(func: Callable[P, Awaitable[None]]) -> Callable[P, Awaitable[None]]:
    @functools.wraps(func)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> None:
        try:
            await func(*args, **kwargs)
        except Exception:
            log.exception("background task failed", extra={"task": func.__name__})
            raise
    return wrapper


@reported
async def send_receipt(order_id: int) -> None:
    ...

Scheduling after commit is the other rule worth enforcing. add_task() before await session.commit() schedules work for a transaction that may still fail; the reload-and-check pattern guards against it, but ordering the calls correctly avoids the question.

Advanced: Keeping Background Work Out of the Request Pool

Every background task that opens a session draws a connection from the same pool as the requests. A burst of requests that each schedule a task doubles connection demand shortly after the burst, exactly when the pool is already busy, and requests start waiting on pool_timeout for connections held by work nobody is waiting for.

Two fixes, which combine well. Cap concurrent background database work with a semaphore, so tasks queue in memory rather than in the pool:

import asyncio

_background_db = asyncio.Semaphore(3)


async def send_receipt(order_id: int) -> None:
    async with _background_db:
        async with Session() as session:
            ...

Or give background work its own small engine, so it cannot take connections from requests at all:

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

background_engine = create_async_engine(
    "postgresql+asyncpg://shop:secret@db/shop",
    pool_size=3,
    max_overflow=0,
    pool_timeout=60,
    connect_args={"server_settings": {"application_name": "shop-api-background"}},
)
BackgroundSession = async_sessionmaker(background_engine, expire_on_commit=False)

A separate engine also makes background load visible in pg_stat_activity under its own application_name, and lets you give it a longer pool_timeout — a receipt that waits a minute is fine; a request that waits a minute is an outage. Dispose both engines in the lifespan shutdown, after in-flight background tasks have had a chance to finish, as described in setting up an async engine from scratch.

Know when to stop using BackgroundTasks altogether. They run in the web process, so a deploy or crash between the response and the task loses the work, and there are no retries. For anything that must happen — a payment capture, an inventory adjustment, a webhook to a partner — use a job queue. arq keeps the async model and the same session factory; Celery is covered in using SQLAlchemy async with Celery task workers. The rule about sessions is unchanged in every case: the task receives identifiers and opens its own session.

Testing Background Tasks That Use the Database

Background tasks are easy to leave untested, because an endpoint test passes as soon as the response comes back. Two kinds of test catch the bugs in this guide.

The first tests the task function directly, as an ordinary coroutine with an ID. Because the task opens its own session from the factory, the test needs that factory to point at the test database — which is the strongest argument for importing Session from one module rather than constructing sessions ad hoc:

import pytest
from sqlalchemy import select

from shop import background
from shop.models import Order


@pytest.mark.asyncio
async def test_send_receipt_marks_order(session_factory, order_factory, outbox):
    order_id = await order_factory(receipt_sent=False)

    await background.send_receipt(order_id)

    async with session_factory() as session:
        order = await session.scalar(select(Order).where(Order.id == order_id))
    assert order.receipt_sent is True
    assert len(outbox) == 1


@pytest.mark.asyncio
async def test_send_receipt_is_idempotent(order_factory, outbox):
    order_id = await order_factory(receipt_sent=False)
    await background.send_receipt(order_id)
    await background.send_receipt(order_id)
    assert len(outbox) == 1

The second is an endpoint test that proves the task runs after the request's session is gone. With httpx.AsyncClient and ASGITransport, background tasks run before the client call returns, so the assertion after the call sees their effect. The test that matters is the one that would have failed with the original bug — the task reading a relationship the endpoint never loaded.

One caveat for suites that roll back a single outer transaction per test, as in rolling back database state between async tests: a task that opens its own session from the application factory opens its own connection, outside the test's transaction, and will not see the test's uncommitted rows. Either bind the factory to the test connection during the test, or commit fixture data and clean up by truncation for this group of tests.

Frequently Asked Questions

Related

• Integrating SQLAlchemy Async with FastAPI and Starlette — The parent guide: request-scoped sessions and dependencies.
• Using expire_on_commit=False in FastAPI dependencies — Why returned objects stay readable after commit.
• Fixing GreenletSpawnError in async SQLAlchemy workflows — The lazy-load failure passed objects trigger.
• Running Concurrent Queries with AsyncSession — One session per unit of concurrent work.