Lesson 5.6 · Databases

Async Database Access

Use an async engine and AsyncSession with async def endpoints - and learn when it really helps. A timing test with a slow query shows the one combination that hurts (a normal session inside async def: 5 requests took 2.03 s instead of 0.5 s), and three async traps: MissingGreenlet on relationships and after commit, and a forgotten await.

Advanced35 min

What you will be able to do

Install and create an async engine
Use AsyncSession in a dependency and in async endpoints
Know which calls need await
Avoid MissingGreenlet with selectinload and refresh
Measure the difference between blocking, threaded and async endpoints
Decide when async database access is worth it

The idea, in plain English

FastAPI runs async def endpoints on one event loop, and def endpoints in a pool of threads. While an async endpoint waits for something - with await - the loop serves other requests. A database query is a wait, so an async driver lets one process handle many requests that are all waiting on the database.

SQLAlchemy and SQLModel have an async version of everything: create_async_engine, AsyncSession, and await on every call that talks to the database. The rules are stricter: nothing may load from the database without await - so relationships and expired attributes need care.

We rebuild the to-do API with async, hit the async traps, and then measure. The result matters more than the code: async is not automatically faster, but mixing a normal session into async def is clearly slower. SQLAlchemy 2.1.4 with greenlet 3.5.6, aiosqlite 0.22.1, SQLModel 0.0.48.

Worked example: The to-do API with an async engine on SQLite, then three versions of one slow endpoint, each hit by 5 requests at once.

workflowFive requests at oncestep 1 / 3

5 requests

at the same time

/blocking

async def + Session

✗

/threads

def + Session

/async

async def + AsyncSession

React Flow

1 - The mistake

A normal session blocks the event loop. The requests wait in a line.

5 at once

2.03 s

Each request runs a query of about 0.4 s. Real timings from our three endpoints.

Words you will see in this lesson

A few async words.

Small dictionary

Event loopThe part that runs async code and switches between tasks while they wait.

await"Wait here, and let other tasks run meanwhile."

Blocking callA call that waits WITHOUT letting other tasks run.

Async driverA database library that works with await: aiosqlite, asyncpg.

greenletA helper SQLAlchemy needs to run its async mode.

MissingGreenletThe error when something tries to load from the database without await.

An everyday example: a restaurant waiter

A good waiter takes an order to the kitchen, and while the food cooks, serves other tables - that is await.
A bad waiter stands at the kitchen door until the food is ready, and every other table waits - a blocking call inside async def.
A restaurant can also hire one waiter per table - that is def endpoints in threads. It works too, it just needs more waiters.

Step 1 - install the async parts

Two packages are needed on top of SQLModel:

aiosqlite - the async driver for SQLite (for PostgreSQL it would be asyncpg).
sqlalchemy[asyncio] - pulls in greenlet. Without it, the first async query failed with an ImportError.
The URL names the driver: sqlite+aiosqlite:///todos.db (for PostgreSQL: postgresql+asyncpg://user:password@host/db).

Output - without greenlet

ImportError: The SQLAlchemy asyncio module requires that the Python 'greenlet' library is installed.  In order to ensure this dependency is available, use the 'sqlalchemy[asyncio]' install target:  'pip install sqlalchemy[asyncio]'

Install

pip install "sqlalchemy[asyncio]" aiosqlite

Example 1 - the async to-do API

Compare main.py with Lesson 5.3:

create_async_engine instead of create_engine.
get_session is async and uses async with AsyncSession(engine) - from sqlmodel.ext.asyncio.session.
create_all is not async, so lifespan runs it with await conn.run_sync(...). It also adds a first user, and closes the engine at shutdown with await engine.dispose().
Every call that talks to the database is awaited: commit, refresh, exec, get.
session.add is NOT awaited - it only marks the object; the database is reached at commit.
The endpoints are async def.

models.py

from sqlmodel import Field, Relationship, SQLModel


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    todos: list["Todo"] = Relationship(back_populates="owner")


class Todo(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    title: str
    done: bool = False
    owner_id: int = Field(foreign_key="user.id")
    owner: User = Relationship(back_populates="todos")


class TodoCreate(SQLModel):
    title: str = Field(min_length=1, max_length=100)
    owner_id: int


class TodoPublic(SQLModel):
    id: int
    title: str
    done: bool
    owner_id: int

Example 1 - main.py

from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.ext.asyncio import create_async_engine
from sqlmodel import SQLModel, select
from sqlmodel.ext.asyncio.session import AsyncSession

from models import Todo, TodoCreate, TodoPublic, User

engine = create_async_engine("sqlite+aiosqlite:///todos.db")    # the async driver in the URL


async def get_session():
    async with AsyncSession(engine) as session:      # async with, not with
        yield session


SessionDep = Annotated[AsyncSession, Depends(get_session, scope="function")]


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)        # create_all is sync: run_sync
    async with AsyncSession(engine) as session:
        if not (await session.exec(select(User))).first():
            session.add(User(name="ravi"))
            await session.commit()
    yield
    await engine.dispose()


app = FastAPI(lifespan=lifespan)


@app.post("/todos", response_model=TodoPublic, status_code=201)
async def create_todo(data: TodoCreate, session: SessionDep):
    todo = Todo.model_validate(data)
    session.add(todo)                                # add is not awaited - it only marks the object
    await session.commit()                           # talks to the database: await
    await session.refresh(todo)
    return todo


@app.get("/todos", response_model=list[TodoPublic])
async def list_todos(session: SessionDep):
    result = await session.exec(select(Todo).order_by(Todo.id))
    return result.all()


@app.get("/todos/{todo_id}", response_model=TodoPublic)
async def read_todo(todo_id: int, session: SessionDep):
    todo = await session.get(Todo, todo_id)
    if todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    return todo

try_async.py

from fastapi.testclient import TestClient

from main import app

with TestClient(app) as c:
    for label, r in [
        ("create", c.post("/todos", json={"title": "Buy milk", "owner_id": 1})),
        ("create", c.post("/todos", json={"title": "Pay rent", "owner_id": 1})),
        ("read 2", c.get("/todos/2")),
        ("read 9", c.get("/todos/9")),
        ("list", c.get("/todos")),
    ]:
        print(f"{label:<7} -> {r.status_code} {r.text}")

Output

create  -> 201 {"id":1,"title":"Buy milk","done":false,"owner_id":1}
create  -> 201 {"id":2,"title":"Pay rent","done":false,"owner_id":1}
read 2  -> 200 {"id":2,"title":"Pay rent","done":false,"owner_id":1}
read 9  -> 404 {"detail":"Todo not found"}
list    -> 200 [{"id":1,"title":"Buy milk","done":false,"owner_id":1},{"id":2,"title":"Pay rent","done":false,"owner_id":1}]

The async traps

traps.py shows what changes compared with a normal session:

1. todo.owner after session.get(Todo, 1): MissingGreenlet. A normal session would load the owner quietly; an async one cannot, because loading needs await.
2. The fix: load it in the query with .options(selectinload(Todo.owner)).
3. user.name right after commit: MissingGreenlet again. Commit expired the object, and reloading needs await.
4. The fix: await session.refresh(user) - or create the session with expire_on_commit=False, which kept the values (user mohan, id 4).
5. Forgetting await: session.exec(...) returned a coroutine, and .all() failed with AttributeError. Python also printed "RuntimeWarning: coroutine ... was never awaited".

traps.py

import asyncio

from sqlalchemy.orm import selectinload
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession

from main import engine
from models import Todo, User


async def main():
    async with AsyncSession(engine) as session:
        # 1. Reading a relationship that is not loaded yet
        todo = await session.get(Todo, 1)
        try:
            print("owner:", todo.owner.name)
        except Exception as e:
            print("1. lazy load:     ", type(e).__name__ + ":", str(e).splitlines()[0][:100])

        # 2. The fix: load it in the query
        result = await session.exec(select(Todo).where(Todo.id == 1).options(selectinload(Todo.owner)))
        print("2. selectinload:  ", result.one().owner.name)

    async with AsyncSession(engine) as session:
        # 3. Reading an attribute after commit
        user = User(name="anu")
        session.add(user)
        await session.commit()
        try:
            print(user.name)
        except Exception as e:
            print("3. after commit:  ", type(e).__name__ + ":", str(e).splitlines()[0][:100])
        await session.refresh(user)                  # the fix: refresh (it awaits the reload)
        print("4. after refresh: ", user.name)

    async with AsyncSession(engine) as session:
        # 5. Forgetting await
        result = session.exec(select(Todo))
        try:
            result.all()
        except Exception as e:
            print("5. no await:      ", type(e).__name__ + ":", e)
        await result                                 # tidy up the coroutine we made


asyncio.run(main())

Output

lazy load:      StatementError: (sqlalchemy.exc.MissingGreenlet) greenlet_spawn has not been called; can't call await_() here. Was I
selectinload:   ravi
after commit:   StatementError: (sqlalchemy.exc.MissingGreenlet) greenlet_spawn has not been called; can't call await_() here. Was I
after refresh:  anu
no await:       AttributeError: 'coroutine' object has no attribute 'all'

Output - AsyncSession(engine, expire_on_commit=False)

expire_on_commit=False, after commit: mohan 4

Example 2 - when does async help?

speed.py runs the same slow query (counting to 3 million in SQL, about 0.4 s) in three endpoints. We ran it with uvicorn and sent 1 request, then 5 at the same time:

/blocking (async def + a normal Session) - 5 requests took 2.03 s: one after another. The normal session blocked the event loop, so nothing else could run - not even other endpoints: a call to /async took 1.38 s instead of 0.39 s while three /blocking requests were running.
/threads (def + a normal Session) - 0.49 s. FastAPI ran each request in a thread, so they overlapped.
/async (async def + AsyncSession) - 0.53 s. They overlapped too.
One request alone took 0.38-0.39 s in all three - async did not make a single query faster.

Example 2 - speed.py

from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine
from sqlmodel import Session, create_engine
from sqlmodel.ext.asyncio.session import AsyncSession
from fastapi import FastAPI

SLOW_SQL = text("""
    WITH RECURSIVE n(x) AS (SELECT 1 UNION ALL SELECT x + 1 FROM n WHERE x < 3000000)
    SELECT COUNT(*) FROM n
""")                                                 # a query that takes about half a second

sync_engine = create_engine("sqlite:///todos.db")
async_engine = create_async_engine("sqlite+aiosqlite:///todos.db")
app = FastAPI()


@app.get("/blocking")
async def blocking():                                # async def + a NORMAL session: the mistake
    with Session(sync_engine) as session:
        return {"count": session.exec(SLOW_SQL).one()[0]}


@app.get("/threads")
def threads():                                       # def + a normal session: runs in a thread
    with Session(sync_engine) as session:
        return {"count": session.exec(SLOW_SQL).one()[0]}


@app.get("/async")
async def use_async():                               # async def + AsyncSession
    async with AsyncSession(async_engine) as session:
        return {"count": (await session.exec(SLOW_SQL)).one()[0]}

bench.sh

for path in blocking threads async; do
  curl -s -o /dev/null localhost:8361/$path                      # warm up
  one=$( { /usr/bin/time -p curl -s -o /dev/null localhost:8361/$path; } 2>&1 | awk '/real/{print $2}')
  start=$(python3 -c 'import time; print(time.time())')
  for i in 1 2 3 4 5; do curl -s -o /dev/null localhost:8361/$path & done; wait
  five=$(python3 -c "import time; print(round(time.time() - $start, 2))")
  printf "/%-9s one request: %ss   5 at the same time: %ss\n" "$path" "$one" "$five"
done

Output

/blocking  one request: 0.38s   5 at the same time: 2.03s
/threads   one request: 0.38s   5 at the same time: 0.49s
/async     one request: 0.39s   5 at the same time: 0.53s

So: async or not?

Never use a normal Session inside async def. That is the only clearly bad choice.
def + Session is simple and fine for most apps - FastAPI gives it threads.
async def + AsyncSession pays off when one server handles very many waiting requests at once, or when the endpoint also awaits other things (HTTP calls, AI models).
Pick one style per project. Async needs more care: await everywhere, selectinload for relationships, refresh after commit.
SQLite is still one file: async does not make it handle many writers. For real concurrency use PostgreSQL (Module 10).

Async SQLModel at a glance

Engine

The driver is in the URL.

create_async_engine("sqlite+aiosqlite:///todos.db")

Session dependency

async with.

async def get_session():
    async with AsyncSession(engine) as session:
        yield session

Create tables

Sync code inside async.

async with engine.begin() as conn:
    await conn.run_sync(SQLModel.metadata.create_all)

Talk to the database

Always await.

await session.commit(); await session.refresh(todo)
(await session.exec(select(Todo))).all()

Relationships

Load in the query.

select(Todo).options(selectinload(Todo.owner))

Try it yourself

The code does not change. Swap the content string and the program does something else entirely.

Block the others

“Start a request to /blocking and, while it runs, call /threads. How long does /threads take?”

More requests

“Change bench.sh to send 20 requests at once. Which endpoint grows fastest?”

Default settings

“Remove scope="function" from SessionDep in main.py. Does anything change for these endpoints?”

Async relationships

“Add GET /users/{id}/todos with selectinload(User.todos). What happens without selectinload?”

What usually goes wrong

A normal Session inside async def

It blocked the event loop: 2.03 s instead of about 0.5 s for 5 requests.

✗ async def f():
    with Session(engine) as s: ...

✓ def f():
    with Session(engine) as s: ...

Reading a relationship lazily

MissingGreenlet. Use selectinload in the query.

Reading attributes after commit

MissingGreenlet. await session.refresh(obj) or expire_on_commit=False.

Forgetting await

'coroutine' object has no attribute 'all'.

✗ session.exec(select(Todo)).all()

✓ (await session.exec(select(Todo))).all()

Installing only aiosqlite

The greenlet ImportError. Install sqlalchemy[asyncio].

Practice

Write these yourself before opening anything. Getting them wrong first is most of how this sticks.

Add async PATCH /todos/{todo_id} and DELETE /todos/{todo_id} to main.py (same rules as Lesson 5.3, including refusing null). Then write GET /todos-with-owner that returns each to-do with its owner’s name, using exactly 2 SQL queries - check with echo=True.

Show hint

await session.delete(todo) is awaited in AsyncSession. For the owner names: select(Todo).options(selectinload(Todo.owner)).

Key points

Async needs an async driver (aiosqlite, asyncpg) and sqlalchemy[asyncio].
AsyncSession: await commit, refresh, exec, get and delete; add is not awaited.
Relationships must be loaded in the query (selectinload).
After commit, refresh - or use expire_on_commit=False.
A normal Session inside async def blocks every request.
def + Session and async def + AsyncSession both handle parallel requests; async pays off with very many waiting requests.

Quick check before you move on

How long did 5 parallel requests take on /blocking, /threads and /async?

2.03 s, 0.49 s and 0.53 s.

Which error did todo.owner give in an AsyncSession?

MissingGreenlet (inside a StatementError).

Which call is NOT awaited: add, commit or refresh?

add.

Which package brings greenlet?

sqlalchemy[asyncio].

Interview questions

Does async make database access faster in FastAPI?

Not per query. It improves concurrency under many simultaneous I/O-bound requests by not tying up a thread per request. Sync endpoints in the threadpool are fine for many apps; the real anti-pattern is blocking calls inside async def.

What is MissingGreenlet and how do you avoid it?

SQLAlchemy raising because implicit I/O (lazy loading, expired attribute refresh) happened outside an awaited context. Use eager loading (selectinload), await session.refresh, expire_on_commit=False, or AsyncAttrs.awaitable_attrs.

Quiz

1.
Why was /blocking slow only when requests came together?
2.
Why can an async session not load todo.owner by itself?
3.
Why does lifespan use run_sync for create_all?

Comments

Loading comments...