asyncpg and AsyncEngine: the driver and the FastAPI patterns

Capsule overview

In capsule 03 you saw the five general parameters of the SQLAlchemy pool. Those parameters work the same with any driver: psycopg sync, psycopg async, asyncpg. But the chosen driver changes details that will matter to you in production.

This capsule focuses specifically on asyncpg + SQLAlchemy 2.0 AsyncEngine — the canonical stack for FastAPI. You're going to see:

• Why asyncpg is the recommended driver (vs psycopg async).
• asyncpg's statement cache: what it is, why it exists, and why it's a problem when you add PgBouncer (capsule 06).
• How create_async_engine and async_sessionmaker connect, and what each relevant parameter does (expire_on_commit, connect_args).
• The canonical session-per-request pattern with Depends(get_db) in FastAPI, which is what avoids most leaks.
• How to close the engine correctly when shutting down the app (lifespan handler).

By the end you'll have an app/db.py module ready for production, with the patterns that avoid the most common errors in async.

Mental model: asyncpg as a specialized engine

If SQLAlchemy is the chassis and transmission, the driver is the engine. Switching from psycopg sync to psycopg async is like going from gasoline to diesel — a different fuel, the same principle. Moving to asyncpg is like switching to a turbo engine designed specifically to run on an async circuit — faster, less overhead, but with its own maintenance (statement cache, specific configuration).

asyncpg was designed from scratch for asyncio. It's not an async wrapper over sync code — it's native async code that speaks PostgreSQL's binary protocol directly. That's why it's the fastest driver for Python async. But that performance comes with details you need to know.

Why asyncpg vs psycopg async

SQLAlchemy 2.0 supports several drivers for PostgreSQL in async:

For a new FastAPI: asyncpg is the recommended default. It's used by Supabase, Litestar, official FastAPI templates.

Technical reason for the performance: asyncpg doesn't use libpq (PostgreSQL's official C library). It implements the binary protocol directly in a C extension + Python. That eliminates parsing, type conversion, and serialization overhead.

Caveat: asyncpg handles some things differently from libpq. The most relevant: prepared statements. It prepares them automatically and caches them. In setups with PgBouncer transaction mode, this breaks — capsule 06 covers the fix.

Base setup: create_async_engine in detail

# app/db.py
from sqlalchemy.ext.asyncio import (
    create_async_engine,
    async_sessionmaker,
    AsyncSession,
)

DATABASE_URL = "postgresql+asyncpg://bookstore:bookstore@localhost:5432/bookstore"

engine = create_async_engine(
    DATABASE_URL,

    # Pool params (capsule 03)
    pool_size=10,
    max_overflow=20,
    pool_timeout=10,
    pool_pre_ping=True,
    pool_recycle=3600,

    # Async-specific
    echo=False,                       # True only in local debugging
    future=True,                      # 2.0 API (default in SQLAlchemy 2.0)
    connect_args={
        "server_settings": {
            "application_name": "bookstore-api",  # appears in pg_stat_activity
            "jit": "off",                          # optional, off for simple queries
        },
        # "statement_cache_size": 0,  # ← We'll add it in capsule 06 with PgBouncer
    },
)

SessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,           # critical in async, see below
    autoflush=False,                  # explicit control of the flush
    autocommit=False,                 # explicit for clarity
)

What each async-relevant parameter does

echo=False

When it's True, SQLAlchemy logs every SQL executed. Useful in local debugging, unviable in production — you multiply the log by 10x minimum and kill performance. If you need temporary SQL logging in production, use auto_explain (module 5) instead of echo.

future=True

Enables SQLAlchemy's 2.0 API. In 2.0+ it's the default, so it's not necessary explicitly. We leave it visible for clarity.

connect_args.server_settings

They pass to the PostgreSQL server at the start of each session. Useful ones:

• application_name: appears in pg_stat_activity and in logs. Essential for distinguishing your app from others connected to the same DB. Put something descriptive.
• jit: if your app does only simple queries (most of FastAPI), turning JIT off ('off') can improve latency (JIT has compilation overhead you only amortize on large queries).
• statement_timeout: aborts queries that take longer than N ms. Useful as a safety net: statement_timeout: '5000' kills queries that exceed 5 seconds.

connect_args.statement_cache_size (capsule 06)

Specific to asyncpg. Default: 100 (caches up to 100 prepared statements per connection). When you use PgBouncer transaction mode, we'll set it to 0 to avoid the broken prepared statements bug.

expire_on_commit=False: why it's critical in async

This async_sessionmaker parameter is the most commonly misunderstood. It's mandatory in async.

What expire_on_commit=True does (default in sync): after a commit, all loaded objects are marked as "expired". The next access to any attribute triggers a SQL query to refresh.

Why that breaks in async:

# Antipattern with expire_on_commit=True
async with SessionLocal() as session:
    book = await session.get(Book, 1)
    book.title = "Updated"
    await session.commit()
    # ← At this point, book is "expired"
    print(book.title)  # ← Tries to lazy-load → MissingGreenlet error in async

After the commit, accessing book.title would require triggering a synchronous SQL query from async code. SQLAlchemy refuses and raises the MissingGreenlet you saw in module 4.

Solution: expire_on_commit=False.

# Correct pattern
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async with SessionLocal() as session:
    book = await session.get(Book, 1)
    book.title = "Updated"
    await session.commit()
    print(book.title)  # ← Works, the objects keep their state post-commit

Caveat: the objects may have "stale" data if another transaction modified the DB in parallel. For cases where you need fresh data, use an explicit await session.refresh(obj).

Practical rule: in async, always expire_on_commit=False. If you need a refresh, do it explicitly.

asyncpg's statement cache

This is the most technical detail of the capsule and the basis for understanding the PgBouncer gotcha (capsule 06).

What prepared statements are

When you run a query with parameters:

SELECT * FROM books WHERE author_id = $1

PostgreSQL can:

Option A — simple execution: parses, plans, and executes each time. Small but constant overhead.

Option B — prepared statement: parses and plans once, saves the plan associated with a name, and reuses the plan in subsequent executions. Faster on queries that repeat.

-- Prepare (happens once)
PREPARE stmt_42 AS SELECT * FROM books WHERE author_id = $1;

-- Execute (happens many times, reusing the plan)
EXECUTE stmt_42(15);
EXECUTE stmt_42(28);

What asyncpg does

asyncpg automatically prepares each distinct query and caches it per connection. The default cache has capacity for 100 statements. Each new execution of the same query uses the cached plan.

# When asyncpg sees this query for the first time:
await conn.fetch("SELECT * FROM books WHERE author_id = $1", 5)
# → PREPARE __asyncpg_stmt_xxxxx__ AS SELECT * FROM books WHERE author_id = $1
# → EXECUTE __asyncpg_stmt_xxxxx__(5)

# When it sees it the second time (same query, different param):
await conn.fetch("SELECT * FROM books WHERE author_id = $1", 12)
# → EXECUTE __asyncpg_stmt_xxxxx__(12)  ← reuses the plan, without re-preparing

Why it matters

Benefit: repeated queries are ~5-15% faster (plan reused).

Problem with PgBouncer transaction mode: PgBouncer reassigns real connections to different clients per transaction. If your client prepared stmt_xxxxx on connection #5 and then PgBouncer hands it connection #7, stmt_xxxxx doesn't exist on connection #7. Error:

asyncpg.exceptions.InvalidSQLStatementNameError:
prepared statement "__asyncpg_stmt_xxxxx__" does not exist

Random, intermittent, hard to reproduce locally.

Fix (capsule 06):

engine = create_async_engine(
    DATABASE_URL,
    connect_args={"statement_cache_size": 0},  # disables the cache
)

Trade-off: you lose the ~5-15% optimization from plan reuse. You gain: your app doesn't break in production behind PgBouncer.

Important note: this fix is only necessary with PgBouncer transaction mode. Without PgBouncer (client directly to PostgreSQL) or with PgBouncer in session mode, leave statement_cache_size at its default (100).

The canonical FastAPI pattern: session-per-request with Depends

This is the pattern that avoids 90% of leaks in production. If you internalize it, most problems disappear.

# app/db.py
from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession

# (engine and SessionLocal defined above)

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    """Dependency that opens and closes a session per request."""
    async with SessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()  # explicit for clarity, the async with also does it

# app/routes/books.py
from fastapi import APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.db import get_db
from app.models import Book

router = APIRouter()


@router.get("/books/{book_id}")
async def get_book(
    book_id: int,
    db: AsyncSession = Depends(get_db),
):
    result = await db.scalar(select(Book).where(Book.id == book_id))
    if not result:
        raise HTTPException(404, "Book not found")
    return result

What this pattern guarantees

1. Each HTTP request has its own session. There's no shared state between requests.
2. The session always closes when the request finishes, even if there's an exception.
3. The connection returns to the pool automatically on closing the session.
4. There's no way to forget to close. The async with does it.

Antipatterns to avoid

Antipattern 1 — global session:

# ❌ NEVER do this in async
session = SessionLocal()  # global, a single one

@app.get("/books")
async def get_books():
    return await session.execute(...)  # shares the session between requests

Why it fails: multiple concurrent requests share the same session, which causes:

• Concurrent execution errors (Already attached to a connection).
• Mixed transaction states (one request commits, another is left with half-finished data).
• Impossible to debug because the behavior depends on the exact temporal order.

Antipattern 2 — opening a session without a context manager:

# ❌ Don't do this
@app.get("/books")
async def get_books():
    session = SessionLocal()
    result = await session.execute(...)
    return result
    # the session never closes → leak

Why it fails: each request opens a connection that is never released. After N requests (where N = pool_size + max_overflow), your pool is saturated forever.

Antipattern 3 — multiple sessions per request without need:

# ❌ Unnecessary
@app.get("/books/{id}")
async def get_book(id: int, db: AsyncSession = Depends(get_db)):
    book = await db.get(Book, id)
    async with SessionLocal() as another_session:  # ← redundant
        author = await another_session.get(Author, book.author_id)
    return {"book": book, "author": author}

Why it fails: it consumes two connections from the pool per request. If you do this in a popular endpoint, it saturates the pool at double the expected rate.

Solution: always use the session injected by Depends(get_db).

Lifespan handler: closing the engine when shutting down the app

FastAPI 0.110+ uses Starlette's lifespan handler for setup/teardown:

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.db import engine


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup (before the yield)
    print("App starting up")
    yield
    # Shutdown (after the yield)
    print("App shutting down, disposing engine")
    await engine.dispose()


app = FastAPI(lifespan=lifespan)

# Your routers
from app.routes import books
app.include_router(books.router)

Why await engine.dispose() matters

engine.dispose() closes all the pool's connections in an orderly way. Without this:

• In development: when you kill the server with Ctrl+C, the connections stay hung until PostgreSQL detects them as dead (it can take minutes).
• In production: a rolling deploy with Kubernetes kills pods. Without dispose, the old pod's connections stay pointing until the timeout. You accumulate zombie connections on each deploy.
• In testing: if you don't close the engine, pytest can hang at the end of the suite because the connections are open.

Rule: always close the engine in the lifespan shutdown.

Complete recommended configuration for production

# app/db.py
from contextlib import asynccontextmanager
from typing import AsyncGenerator

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

DATABASE_URL = "postgresql+asyncpg://bookstore:bookstore@db:5432/bookstore"

# Engine: a single instance for the whole app
engine = create_async_engine(
    DATABASE_URL,
    # Pool
    pool_size=10,
    max_overflow=20,
    pool_timeout=10,
    pool_pre_ping=True,
    pool_recycle=3600,
    # Async / asyncpg
    echo=False,
    connect_args={
        "server_settings": {
            "application_name": "bookstore-api",
            "statement_timeout": "10000",  # kills queries longer than 10s
        },
        # statement_cache_size is added when you introduce PgBouncer (capsule 06)
    },
)

SessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autoflush=False,
)


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with SessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()


@asynccontextmanager
async def lifespan(app: FastAPI):
    # startup
    yield
    # shutdown
    await engine.dispose()

This snippet is ~50 lines and is the base for any production FastAPI API. Capsule 07 extends it with pool monitoring.

Why this matters in real work

1. asyncpg + AsyncEngine is the canonical stack. Official FastAPI templates (full-stack-fastapi-postgresql, fastapi-best-practices), Litestar starters, and the Supabase backend use exactly this setup. Knowing it is what differentiates a dev who "knows FastAPI" from one who "knows production FastAPI".

2. expire_on_commit=False is the #1 cause of MissingGreenlet. If you leave the default, you'll have cryptic errors after every commit. Knowing it from the start saves you hours of debugging.

3. The session-per-request pattern with Depends is a prerequisite for scaling. Without it, every "the API hangs" bug ends up being a leak. With it, debugging focuses on slow queries, not connection management.

4. asyncpg's statement cache is information that doesn't appear in blogs. Until you introduce PgBouncer and it breaks, nobody is going to explain it to you. Knowing it beforehand (and understanding why the fix is statement_cache_size=0) saves you the subtlest incident that can happen to you.

5. The lifespan handler is one of the first things they check in a senior code review. "Do you close the engine on shutdown?" If you don't, there's a comment on every PR.

Traps and common mistakes

Mistake 1 (conceptual): leaving expire_on_commit=True in async

Symptom: after await session.commit(), accessing any attribute of the object raises MissingGreenlet.

Why it happens: the sync default of expire_on_commit=True means that post-commit all objects are "expired" — the next access triggers a lazy load. In async that's the error you saw in module 4.

How to distinguish: the error comes after the commit, not before. If the pattern is await commit() → access → error, this is it.

How to fix it:

SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

For cases where you need fresh data post-commit, use an explicit await session.refresh(obj).

Mistake 2 (operational): not calling await engine.dispose() on shutdown

Symptom: "Every time I restart the app, pg_stat_activity shows connections from the previous version for several minutes."

Why it happens: without dispose(), the connections stay open until the TCP keepalive detects them as dead (typically 2-5 minutes).

How to distinguish: after a server restart, pg_stat_activity shows application_name = bookstore-api with a state_change before the restart.

How to fix it: use a lifespan handler with await engine.dispose() on shutdown. If you use Gunicorn/uvicorn with worker recycling, make sure the lifespan fires in each worker (generally it does).

Mistake 3 (conceptual): sharing a session between requests

Symptom: intermittent errors like This Session is currently in 'committed' state, no further SQL can be emitted or strange results where one request sees another's data.

Why it happens: some design accidentally shares a global session. The concurrent requests overwrite each other's session state.

How to distinguish: the errors are intermittent and depend on the request order. Reproducing locally with a single request is hard; it appears under load.

How to fix it: always use Depends(get_db). One session per request, no exceptions. If you need long transactional operations that don't fit in a request, use background workers (Celery, ARQ) with their own session, not the HTTP request's.

Mistake 4 (operational): not setting application_name

Symptom: when investigating pg_stat_activity during an incident, you can't distinguish your API's connections from other apps connected to the same DB.

Why it happens: without application_name configured, pg_stat_activity shows generic values like psql, python, or empty.

How to distinguish: query SELECT application_name, count(*) FROM pg_stat_activity GROUP BY application_name. If you see many rows with an empty or generic application_name, this is it.

How to fix it: always set application_name in connect_args:

connect_args={
    "server_settings": {"application_name": "bookstore-api"},
}

For advanced cases, you can include per-instance info: f"bookstore-api-{INSTANCE_ID}". Useful for distinguishing connections from different pods in Kubernetes.

Mistake 5 (conceptual): assuming asyncpg "has no cache" because it's async

Symptom: you enable PgBouncer transaction mode expecting "everything to work" because your app is async. After a few minutes, intermittent prepared statement does not exist errors begin.

Why it happens: asyncpg caches prepared statements automatically, independent of async. PgBouncer transaction mode reassigns connections, breaking the cache.

How to distinguish: the specific error asyncpg.exceptions.InvalidSQLStatementNameError: prepared statement "__asyncpg_stmt_xxxxx__" does not exist. It appears intermittently, not on every request.

How to fix it: capsule 06 covers this in detail. Spoiler: connect_args={"statement_cache_size": 0} when you use PgBouncer transaction mode.

Exercises

Exercise 1: complete setup from scratch

Create an app/db.py for a new FastAPI API following the canonical pattern. Include the engine, sessionmaker, get_db dependency, and lifespan handler.

# app/db.py
from contextlib import asynccontextmanager
from typing import AsyncGenerator

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

DATABASE_URL = "postgresql+asyncpg://bookstore:bookstore@localhost:5432/bookstore"

engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_timeout=10,
    pool_pre_ping=True,
    pool_recycle=3600,
    echo=False,
    connect_args={
        "server_settings": {
            "application_name": "bookstore-api",
            "statement_timeout": "10000",
        },
    },
)

SessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autoflush=False,
)


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with SessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    await engine.dispose()

# app/main.py
from fastapi import FastAPI
from app.db import lifespan
from app.routes import books

app = FastAPI(lifespan=lifespan)
app.include_router(books.router)

# app/routes/books.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.db import get_db
from app.models import Book

router = APIRouter(prefix="/books", tags=["books"])


@router.get("/{book_id}")
async def get_book(
    book_id: int,
    db: AsyncSession = Depends(get_db),
):
    book = await db.scalar(select(Book).where(Book.id == book_id))
    if not book:
        raise HTTPException(404, "Book not found")
    return book

uvicorn app.main:app --reload

SELECT application_name, count(*)
FROM pg_stat_activity
WHERE application_name = 'bookstore-api'
GROUP BY application_name;

Exercise 2: reproduce and fix MissingGreenlet

Write an endpoint that reproduces MissingGreenlet by using expire_on_commit=True. Then fix it.

# app/routes/buggy.py
from fastapi import APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

DATABASE_URL = "postgresql+asyncpg://bookstore:bookstore@localhost:5432/bookstore"

# Engine and session with expire_on_commit=True (default antipattern)
engine_buggy = create_async_engine(DATABASE_URL)
SessionBuggy = async_sessionmaker(engine_buggy, expire_on_commit=True)

# Engine and session with expire_on_commit=False (correct)
engine_ok = create_async_engine(DATABASE_URL)
SessionOk = async_sessionmaker(engine_ok, expire_on_commit=False)

router = APIRouter()


@router.post("/books/buggy")
async def update_buggy():
    async with SessionBuggy() as session:
        # Simulate an update
        await session.execute(text("UPDATE books SET title = title WHERE id = 1"))
        await session.commit()

        # Load after the commit
        result = await session.scalar(select(Book).where(Book.id == 1))
        # This will break on the next access to an attribute:
        return {"title": result.title}  # ← MissingGreenlet here


@router.post("/books/correct")
async def update_correct():
    async with SessionOk() as session:
        await session.execute(text("UPDATE books SET title = title WHERE id = 1"))
        await session.commit()

        result = await session.scalar(select(Book).where(Book.id == 1))
        return {"title": result.title}  # Works

curl -X POST http://localhost:8000/books/buggy
# 500 Internal Server Error - MissingGreenlet in the logs

curl -X POST http://localhost:8000/books/correct
# {"title": "Some Book Title"}

Exercise 3: identify a leak from an unclosed session

You have this endpoint and you notice that after a few requests, the pool saturates. Identify the bug and fix it.

@app.get("/books/{book_id}")
async def get_book(book_id: int):
    session = SessionLocal()
    book = await session.scalar(select(Book).where(Book.id == book_id))
    return book

for i in {1..30}; do curl -s http://localhost:8000/books/1 > /dev/null; done

docker exec -it bookstore-pg psql -U bookstore -d bookstore -c "
SELECT count(*), state FROM pg_stat_activity
WHERE application_name = 'bookstore-api' GROUP BY state;
"

@app.get("/books/{book_id}")
async def get_book(book_id: int, db: AsyncSession = Depends(get_db)):
    book = await db.scalar(select(Book).where(Book.id == book_id))
    return book

@app.get("/books/{book_id}")
async def get_book(book_id: int):
    async with SessionLocal() as session:
        book = await session.scalar(select(Book).where(Book.id == book_id))
        return book

Exercise 4: configure application_name and verify in pg_stat_activity

Configure your engine so it appears with application_name = "my-bookstore-prod" in PostgreSQL. Verify with pg_stat_activity.

# app/db.py
import os

INSTANCE_ID = os.environ.get("INSTANCE_ID", "local")

engine = create_async_engine(
    DATABASE_URL,
    connect_args={
        "server_settings": {
            "application_name": f"my-bookstore-prod-{INSTANCE_ID}",
        },
    },
)

INSTANCE_ID=worker-1 uvicorn app.main:app

curl http://localhost:8000/books/1

docker exec -it bookstore-pg psql -U bookstore -d bookstore -c "
SELECT application_name, count(*), state
FROM pg_stat_activity
WHERE application_name LIKE 'my-bookstore-prod%'
GROUP BY application_name, state;
"

       application_name        | count | state
-------------------------------+-------+--------
 my-bookstore-prod-worker-1   |    10 | idle
 my-bookstore-prod-worker-1   |     1 | active

# In each pod
INSTANCE_ID=$HOSTNAME uvicorn app.main:app

# Then, the query shows:
SELECT application_name, count(*) FROM pg_stat_activity
WHERE application_name LIKE 'my-bookstore-prod%'
GROUP BY application_name;

       application_name              | count
-------------------------------------+-------
 my-bookstore-prod-pod-abc123        |    10
 my-bookstore-prod-pod-def456        |    10
 my-bookstore-prod-pod-ghi789        |    10

Exercise 5: test the effect of asyncpg's statement cache

Connect directly with asyncpg (without SQLAlchemy) and verify that repeated queries are cached. Then test with statement_cache_size=0 and note the difference.

# bench_stmt_cache.py
import asyncio
import time
import asyncpg

DSN = "postgresql://bookstore:bookstore@localhost:5432/bookstore"


async def benchmark(stmt_cache_size: int):
    conn = await asyncpg.connect(DSN, statement_cache_size=stmt_cache_size)
    # Warm up: run the query once
    await conn.fetch("SELECT * FROM books WHERE id = $1", 1)

    # Measure 1000 executions
    start = time.perf_counter()
    for i in range(1000):
        await conn.fetch("SELECT * FROM books WHERE id = $1", i % 100 + 1)
    elapsed = time.perf_counter() - start
    await conn.close()
    return elapsed


async def main():
    cached = await benchmark(stmt_cache_size=100)
    no_cache = await benchmark(stmt_cache_size=0)

    print(f"With cache (default): {cached*1000:.1f}ms total ({cached:.2f}ms per query)")
    print(f"Without cache:        {no_cache*1000:.1f}ms total ({no_cache:.2f}ms per query)")
    print(f"Difference: {((no_cache - cached) / cached) * 100:.1f}% slower without cache")


asyncio.run(main())

With cache (default): 320.5ms total (0.32ms per query)
Without cache:        385.2ms total (0.39ms per query)
Difference: 20.2% slower without cache

Summary and next step

In this capsule you:

• Confirmed asyncpg as the canonical driver for FastAPI with SQLAlchemy 2.0.
• Learned the complete setup of create_async_engine, async_sessionmaker, and the get_db dependency.
• Internalized expire_on_commit=False as mandatory in async.
• Learned about asyncpg's statement cache and why it matters (a hook to capsule 06).
• Adopted the session-per-request pattern with Depends(get_db).
• Configured a lifespan handler with await engine.dispose() for a clean shutdown.

Before moving on, you should be able to:

• Create app/db.py from scratch following the canonical pattern, without copying it.
• Diagnose MissingGreenlet and know that the fix is expire_on_commit=False.
• Identify connection leaks from unclosed sessions in code reviews.
• Set application_name so your connections are identifiable in pg_stat_activity.

Next capsule — PgBouncer fundamentals. So far all the pooling lives in the client (SQLAlchemy + asyncpg). In the next capsule we introduce PgBouncer: the external pool that sits between your app and PostgreSQL. You're going to see why it's necessary at a certain scale (remember the "4 instances × pool_size > max_connections" calculation from capsule 02), how it works architecturally, and the three pooling modes (session/transaction/statement) with the complete decision matrix among them. That capsule and 06 are the most technically loaded of the module.

Resources

1. SQLAlchemy 2.0 — AsyncEngine — the complete official async reference.
2. asyncpg documentation — the official driver.
3. asyncpg — Statement Cache — the specific section on the prepared statements cache.
4. FastAPI — SQLAlchemy 2.0 + asyncpg tutorial — canonical patterns.
5. Mike Bayer — "What's New in SQLAlchemy 2.0" — relevant changes in async.
6. full-stack-fastapi-postgresql template — a reference setup with asyncpg.

Module 6 — Database Performance & Query Tuning Guide