Async and Databases

7 pages · ~5 min✓ Reviewed

Builds on Dependency Injection. Next up: APIRouter and Project Structure.

FastAPIPage 1

async def vs def Endpoints

1. Where each runs

async def runs directly on the event loop
def is sent to a threadpool (anyio, 40 threads default)
Rule: async def only if every await is async-native
Blocking libs (requests, psycopg2, time.sleep) → plain def

	async def	def
Concurrency	Cheap	Limited by thread count
Blocking call	Stalls ALL requests	Isolated per request

2. Example

/a and /b stay responsive; /c freezes the server

python

@app.get('/a')
async def a(): await asyncio.sleep(1)
@app.get('/b')
def b(): time.sleep(1)
@app.get('/c')
async def c(): time.sleep(1)  # bad

3. Mistake

Blocking code inside async def: requests.get, sync DB driver, CPU loops

Fix

Use plain def, or await run_in_threadpool(fn) / await asyncio.to_thread(fn).

FastAPIPage 2

SQLAlchemy 2.0 Async Engine and Session

1. Engine and Factory

create_async_engine('postgresql+asyncpg://...') needs an async driver
Drivers: asyncpg, aiosqlite; a sync driver URL fails
async_sessionmaker(engine, expire_on_commit=False) is the session factory
expire_on_commit=False: attributes are not expired after commit
Why: an expired attribute reloads lazily, which is illegal IO in async

2. Queries and Writes

2.0 style: await session.execute(select(...)), then a result method
Write: add, await commit(), then await refresh(obj) for server-generated fields

python

async def demo(session):
    r = await session.execute(select(User).where(User.id == 1))
    user = r.scalar_one_or_none()
    obj = User(name="Ada")
    session.add(obj)
    await session.commit()
    await session.refresh(obj)

fragment: needs a live AsyncSession

3. MissingGreenlet

FastAPIPage 3

Session per Request via Dependencies

1. The Pattern

One AsyncSession per request, created in a dependency
get_db is an async generator; code after yield is cleanup
Inject: db: Annotated[AsyncSession, Depends(get_db)]
Sub-dependencies in one request share the same session (dependency cache)

Request
Open sessionasync with
Endpoint runs
Exception?rollback
Closealways

python

async def get_db():
    async with SessionLocal() as session:
        try:
            yield session
        except Exception:
            await session.rollback()
            raise

rollback only if an exception propagates

2. Transactions

Commit is explicit: await session.commit() in endpoint or service

FastAPIPage 4

Background Tasks

1. What It Does

background_tasks.add_task(send_email, to, body)
Runs after the response is sent
def task → threadpool; async def task → event loop
Args after the function are passed to add_task

Endpointadd_task
Response sentclient done
Task runsafter

2. Task Runner or Queue

	BackgroundTasks	Celery / ARQ / RQ
Use for	Short fire-and-forget: emails, logs	Heavy or durable jobs
Process dies	Task lost	Job survives
Retries	None	Built in

FastAPIPage 5

Lifespan Events

1. What lifespan is

lifespan is an @asynccontextmanager passed as FastAPI(lifespan=lifespan)
Code before yield runs at startup
Code after yield runs at shutdown

2. Startup and shutdown work

python

@asynccontextmanager
async def lifespan(app: FastAPI):
    engine = create_async_engine(DB_URL)
    app.state.engine = engine
    app.state.SessionLocal = async_sessionmaker(
        engine, expire_on_commit=False)
    yield
    await engine.dispose()

engine built once, disposed once

Phase	Typical work
Startup	async engine + sessionmaker, load ML models, Redis or HTTP client pools
Store on	`app.state`
Shutdown	`await engine.dispose()`, close clients, flush resources

3. Old way and mistakes

FastAPIPage 6

FastAPI Async Cheat Sheet

1. Blocking code

	async def	def
Runs on	event loop	threadpool
Blocking call	stalls ALL requests	safe

Blocking lib? Use def
Or asyncio.to_thread / run_in_threadpool inside async def

2. Engine and lifespan

python

@asynccontextmanager
async def lifespan(app):
    engine = create_async_engine(URL)
    app.state.SessionLocal = async_sessionmaker(
        engine, expire_on_commit=False)
    yield
    await engine.dispose()

FastAPI(lifespan=lifespan)

Engine once, never per request
selectinload avoids MissingGreenlet

3. Session and background

FastAPIPage 7

Check yourself

Two clients call this at once. How long until the second one gets a reply?

About 2s, not 1s.
time.sleep blocks the event loop, so every request waits.
Fix: plain def (threadpool) or await asyncio.sleep(1).

@app.get('/slow')
async def slow():
    time.sleep(1)
    return 'ok'

What goes wrong on the last line?

After commit, loaded attributes are expired.
Touching one on an AsyncSession triggers lazy IO and raises MissingGreenlet.
Fix: expire_on_commit=False or await session.refresh(user).

await session.commit()
print(user.name)

Why does the background task fail?

The task runs after the response, when get_db has already closed db.
Open a new session inside the task.

async def post(bg: BackgroundTasks, db: Annotated[AsyncSession, Depends(get_db)]):
    bg.add_task(audit, db)

Main	`selectinload()` / `joinedload()` in the query
Fallback	`await session.run_sync(fn)` or `awaitable_attrs`

Async and Databases

async def vs def Endpoints

1. Where each runs

2. Example

3. Mistake

SQLAlchemy 2.0 Async Engine and Session

1. Engine and Factory

2. Queries and Writes

3. MissingGreenlet

Session per Request via Dependencies

1. The Pattern

2. Transactions

Background Tasks

1. What It Does

2. Task Runner or Queue

Lifespan Events

1. What lifespan is

2. Startup and shutdown work

3. Old way and mistakes

FastAPI Async Cheat Sheet

1. Blocking code

2. Engine and lifespan

3. Session and background

Check yourself

4. Interview Point

3. Mistakes

3. Example and Mistake