PgBouncer + asyncpg: the prepared statements gotcha and other minor ones

Capsule overview

You just brought up PgBouncer (capsule 05) and migrated your app to point at it in session mode. Everything works — but session mode doesn't leverage PgBouncer. The recommended mode is transaction, and when you change to it the most expensive bug in this whole guide appears:

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

Random. Intermittent. Not reproducible locally. It appears in production at a certain throughput. If you don't know the fix, you debug for weeks.

This capsule teaches you:

• Why the bug happens technically (asyncpg statement cache + PgBouncer transaction mode interaction).
• The exact fix: statement_cache_size=0 in connect_args. One line of configuration.
• Other minor gotchas that appear in transaction mode: SET LOCAL, advisory locks, connections that change.
• How to verify that the fix works: repeated load tests without intermittent errors.

By the end you'll have your FastAPI + asyncpg app running in transaction mode with the correct configuration, and you'll understand why each line is necessary.

Mental model: bank tellers with notes taped to their desk

Imagine a bank with several tellers (real connections to PostgreSQL). When a client (asyncpg) arrives for the first time with a common operation — "deposit to account X" — the teller tapes a note to their desk: "Client A deposits to account X, I already know the format". Next time Client A comes, the teller just says "deposit to X" and everything goes faster — the note is there.

That's prepared statements: query plans cached per connection.

Now you introduce a manager (PgBouncer transaction mode) that assigns clients to tellers dynamically. The manager decides: "Client A, your next deposit will be handled by teller #7". But Client A's note was taped to teller #5's desk.

Client A arrives, says "deposit to X", teller #7 replies "I don't understand, I don't have that note". It fails.

That's the exact bug: asyncpg caches statements per connection, PgBouncer reassigns connections per transaction, the caches don't migrate.

Solution from the analogy: tell Client A "don't tape notes — explain everything each time". They lose a pinch of efficiency (re-explaining) but it works with any teller.

That's statement_cache_size=0: disabling asyncpg's cache.

The bug in detail: how it manifests

Minimal reproduction

Setup: PgBouncer in transaction mode (capsule 05) + asyncpg with default configuration (cache enabled).

# bug_demo.py
import asyncio
import asyncpg

async def main():
    # Connect to PgBouncer (port 6432)
    pool = await asyncpg.create_pool(
        "postgresql://bookstore:bookstore@localhost:6432/bookstore",
        min_size=5,
        max_size=10,
    )

    async def worker(idx):
        for i in range(50):
            try:
                async with pool.acquire() as conn:
                    # This query will be cached as a prepared statement
                    result = await conn.fetch(
                        "SELECT * FROM books WHERE id = $1", i % 100 + 1
                    )
            except asyncpg.exceptions.InvalidSQLStatementNameError as e:
                print(f"[Worker {idx}] FAILED at iter {i}: {e}")
                return

    # Launch many workers in parallel
    await asyncio.gather(*[worker(i) for i in range(20)])
    await pool.close()

asyncio.run(main())

Expected output (after a few iterations):

[Worker 3] FAILED at iter 12: prepared statement "__asyncpg_stmt_3__" does not exist
[Worker 7] FAILED at iter 8: prepared statement "__asyncpg_stmt_5__" does not exist
[Worker 1] FAILED at iter 22: prepared statement "__asyncpg_stmt_8__" does not exist

Why it happens:

1. asyncpg opens connections to PgBouncer's pool.
2. Each client connection internally receives a real PgBouncer connection (transaction mode = assigned per transaction).
3. asyncpg sends PREPARE __asyncpg_stmt_3__ AS SELECT * FROM books WHERE id = $1 on real connection #5.
4. PgBouncer registers in its understanding that "real connection #5 has this prepared statement".
5. The transaction finishes. PgBouncer releases real connection #5.
6. The same client's next query arrives. PgBouncer assigns it real connection #7 (not #5).
7. asyncpg sends EXECUTE __asyncpg_stmt_3__(15).
8. Real connection #7 doesn't have that prepared statement → error.

Why it's random: it depends on how PgBouncer assigns real connections. With little load, you almost always get the same one → it doesn't fail. With a lot of load, the recycles are frequent → it fails often.

Why it doesn't reproduce locally: locally with low throughput you rarely hit a recycle. In production with 200 RPS, the recycles happen every few milliseconds.

The fix: statement_cache_size=0

One line in the configuration. It tells asyncpg: "don't cache prepared statements, run everything as a direct query each time".

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    "postgresql+asyncpg://bookstore:bookstore@localhost:6432/bookstore",
    pool_size=20,
    max_overflow=0,
    pool_timeout=10,
    pool_pre_ping=True,
    pool_recycle=3600,
    connect_args={
        "statement_cache_size": 0,                    # ← THIS IS THE FIX
        "prepared_statement_cache_size": 0,           # ← (optional, see below)
        "server_settings": {
            "application_name": "bookstore-api",
        },
    },
)

What it does exactly

statement_cache_size=0 tells asyncpg not to use prepared statements at all. Each query is sent to the server as a simple parameterized statement:

Without the fix (cache enabled):
  PREPARE __asyncpg_stmt_3__ AS SELECT * FROM books WHERE id = $1
  EXECUTE __asyncpg_stmt_3__(5)

With the fix (cache disabled):
  -- Direct, without a cached name
  SELECT * FROM books WHERE id = $1  -- with parameter 5

Internally asyncpg still uses PostgreSQL's extended protocol (parsing + binding + execute), but it doesn't name or cache the statements. Each execution is independent.

The additional parameter: prepared_statement_cache_size

There's another related parameter, specific to SQLAlchemy (not asyncpg):

connect_args={
    "statement_cache_size": 0,                # from asyncpg
    "prepared_statement_cache_size": 0,       # from SQLAlchemy (its own cache on top of asyncpg)
}

SQLAlchemy can also keep its own plan cache at the ORM level. Although less problematic than asyncpg's with PgBouncer, setting it to 0 is defensive. Most sources recommend it together.

Verification: the bug no longer appears

With the fix, the bug reproduction script should run 1000 iterations without errors:

# bug_fixed_demo.py
import asyncio
import asyncpg

async def main():
    pool = await asyncpg.create_pool(
        "postgresql://bookstore:bookstore@localhost:6432/bookstore",
        min_size=5,
        max_size=10,
        statement_cache_size=0,  # ← FIX
    )
    # ... rest of the code the same ...

No more intermittent failures.

Cost of the fix

It's not free. Disabling the cache costs ~5-15% overhead per query.

Reasons:

• PostgreSQL parses + plans each time (with the cache, it does it once).
• asyncpg can't optimize reuse between calls.
• Each query pays the parse + bind + execute round-trip (vs only execute with the cache).

Real measurement:

# bench_cache.py
import asyncio
import time
import asyncpg

DSN = "postgresql://bookstore:bookstore@localhost:5432/bookstore"  # PG directly
N = 1000

async def benchmark(cache_size):
    conn = await asyncpg.connect(DSN, statement_cache_size=cache_size)
    # warmup
    await conn.fetch("SELECT * FROM books WHERE id = $1", 1)
    start = time.perf_counter()
    for i in range(N):
        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(100)
    no_cache = await benchmark(0)
    print(f"With cache: {cached*1000:.1f}ms ({cached/N*1000:.3f}ms/query)")
    print(f"Without cache: {no_cache*1000:.1f}ms ({no_cache/N*1000:.3f}ms/query)")
    print(f"Overhead: {(no_cache-cached)/cached*100:.1f}%")

asyncio.run(main())

Typical output:

With cache: 320.5ms (0.320ms/query)
Without cache: 385.2ms (0.385ms/query)
Overhead: 20.2%

Is it worth it? Yes in almost all cases. Reasons:

1. 20% of 0.3ms is 0.06ms. Imperceptible to users.
2. The benefit of PgBouncer transaction mode (multiplexing, scaling to more instances without touching PG) is much greater.
3. Without the fix, you can't use transaction mode reliably, you lose all of PgBouncer's benefit.

Exception: apps with extreme low latency (HFT-style, p99 < 5ms). Consider session mode (no overhead but less multiplexing) or not using PgBouncer.

Other gotchas in transaction mode

statement_cache_size=0 solves bug #1 but there are others more subtle. We list the important ones:

Gotcha 2: SET LOCAL only lasts the transaction

Problem:

# In your multitenant app
async with SessionLocal() as session:
    await session.execute(text("SET search_path TO tenant_42"))  # ← without LOCAL, without BEGIN
    # the session may implicitly take different connections for subsequent queries
    result = await session.scalar(select(User).where(User.id == 1))
    # does this query use search_path tenant_42 or the default?

In transaction mode, the answer is non-deterministic. PgBouncer may have changed the real connection between the SET and the SELECT.

Fix: wrap it explicitly in a transaction:

async with SessionLocal() as session:
    async with session.begin():  # explicit transaction
        await session.execute(text("SET LOCAL search_path TO tenant_42"))
        result = await session.scalar(select(User).where(User.id == 1))
        # COMMIT on exiting the with

SET LOCAL inside BEGIN ... COMMIT applies to all the queries of that specific transaction.

Gotcha 3: LISTEN/NOTIFY doesn't work

Problem: LISTEN requires a persistent connection. PgBouncer transaction mode reassigns connections, so the LISTEN stays on a connection that's no longer assigned to the client.

# Antipattern in transaction mode
async with engine.connect() as conn:
    await conn.execute(text("LISTEN test_channel"))
    # Wait for notifications...
    # ← Nothing arrives because the real connection changed.

Fix: use a direct connection to PostgreSQL (not PgBouncer) for listeners:

# A separate engine only for listeners, connected directly to PG
engine_listen = create_async_engine(
    "postgresql+asyncpg://bookstore:bookstore@localhost:5432/bookstore",  # PG directly (port 5432)
    pool_size=2,  # few listeners typically
)

# For the rest of the app, keep PgBouncer on port 6432
engine_app = create_async_engine(
    "postgresql+asyncpg://bookstore:bookstore@localhost:6432/bookstore",  # PgBouncer (port 6432)
    ...
)

Gotcha 4: session advisory locks

Problem: pg_advisory_lock(id) holds the lock until the session ends. In transaction mode, "session" is a transaction.

# Antipattern
async with engine.connect() as conn:
    await conn.execute(text("SELECT pg_advisory_lock(42)"))
    # Do work in another connection while this "should" hold the lock
    # ← The lock is released when the implicit transaction ends.

Fix: use pg_advisory_xact_lock (transaction-level lock) instead of pg_advisory_lock:

async with session.begin():
    await session.execute(text("SELECT pg_advisory_xact_lock(42)"))
    # Lock active during this entire transaction
    # Other transactions wait
    # ...
# COMMIT releases the lock automatically

pg_advisory_xact_lock is meant exactly for this case.

Gotcha 5: cursors WITH HOLD

Problem: DECLARE ... WITH HOLD cursors keep state between transactions. Transaction mode breaks this.

Fix: avoid them. Use pagination (classic cursor pagination, OFFSET/LIMIT with indexes) instead. Guide #13 covers cursor pagination correctly without using WITH HOLD.

Gotcha 6: temp tables that persist

Problem: CREATE TEMP TABLE creates a table visible only in the session. In transaction mode, the "session" is the transaction.

BEGIN;
CREATE TEMP TABLE my_temp (...);
INSERT INTO my_temp ...;
COMMIT;

-- In the next transaction (may be a different real connection):
SELECT * FROM my_temp;  -- ERROR: relation "my_temp" does not exist

Fix: create and use the temp table within the same transaction:

async with session.begin():
    await session.execute(text("CREATE TEMP TABLE my_temp ..."))
    await session.execute(text("INSERT INTO my_temp ..."))
    result = await session.execute(text("SELECT * FROM my_temp"))
# COMMIT, the temp table disappears. OK because you already used the results.

Complete recommended configuration

Combining everything (capsules 03, 04, 05, 06):

# app/db.py — FINAL CONFIGURATION for FastAPI + asyncpg + PgBouncer transaction mode
from contextlib import asynccontextmanager
from typing import AsyncGenerator

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

# Point at PgBouncer (port 6432), not PG directly (5432)
DATABASE_URL = "postgresql+asyncpg://bookstore:bookstore@pgbouncer:6432/bookstore"

engine = create_async_engine(
    DATABASE_URL,

    # Client pool (capsule 03)
    pool_size=20,
    max_overflow=0,             # PgBouncer absorbs peaks on the server side
    pool_timeout=10,
    pool_pre_ping=True,
    pool_recycle=3600,

    # Async / asyncpg (capsules 04 + 06)
    echo=False,
    connect_args={
        "statement_cache_size": 0,             # ← FIX for PgBouncer transaction mode
        "prepared_statement_cache_size": 0,    # ← defensive, SQLAlchemy layer
        "server_settings": {
            "application_name": "bookstore-api",
            "statement_timeout": "10000",      # kills queries longer than 10s
        },
    },
)

SessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,     # capsule 04: critical in async
    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()

This configuration is the one module 8 (final project) uses. If you understand it line by line, you've already internalized this whole module.

Why this matters in real work

1. It's the #1 bug that takes down apps when introducing PgBouncer. Every team that migrates to PgBouncer transaction mode with asyncpg, without knowing the fix, discovers it in production. You already know it beforehand.

2. One line of configuration prevents a postmortem. statement_cache_size=0 is 30 characters. Without it, it's days of "intermittent" debugging until someone finds the right GitHub thread.

3. The secondary gotchas (LISTEN/NOTIFY, SET LOCAL, advisory locks) appear in real applications. Multitenancy with search_path is very common — knowing that it breaks without LOCAL in transaction mode saves you expensive refactors.

4. The knowledge transfers to managed setups. AWS RDS Proxy, Supabase Pooler, Neon Pooler — they all use PgBouncer transaction mode internally. The fix is the same. Knowing it applies to any managed service.

5. It's one of the few things not clearly documented in SQLAlchemy/FastAPI tutorials. SQLAlchemy's official docs mention "consider statement_cache_size for PgBouncer compatibility" in a side note. asyncpg's mentions it but doesn't emphasize it. Popular tutorials omit it. That's why it's bug #1: the documentation doesn't shout it loudly enough.

Traps and common mistakes

Mistake 1 (operational): enabling transaction mode without the fix

Symptom: after changing POOL_MODE: transaction in docker-compose, intermittent errors start to appear:

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

You reproduce locally with low throughput → it doesn't fail. You go to staging with real load → it starts failing past a certain RPS.

Why it happens: you forgot to add statement_cache_size=0 to connect_args.

How to distinguish: the error message is very specific (prepared statement "__asyncpg_stmt_xxx__"). If you see it after migrating to PgBouncer transaction mode, it's almost certainly this.

How to fix it: add statement_cache_size=0 to connect_args in create_async_engine. Restart the app. Verify with sustained load tests.

Mistake 2 (conceptual): thinking the fix degrades performance significantly

Symptom: "I read that statement_cache_size=0 adds 20% overhead. I'll keep the cache and deal with the errors."

Why it happens: confusion between relative and absolute overhead. 20% of 0.3ms = 0.06ms — invisible to humans.

How to distinguish: measure for real (the benchmark exercise from capsule 04). On simple queries, the overhead is <0.1ms. On queries that take 50ms+, the overhead is <1%.

How to fix it: accept the overhead. The benefit of PgBouncer transaction mode (horizontal scalability, RAM in PG, connection capacity) is much greater.

Mistake 3 (conceptual): applying the fix even without PgBouncer

Symptom: someone reads this module, copies the fix to a setup that doesn't use PgBouncer, and wonders why the app is 20% slower.

Why it happens: statement_cache_size=0 is only necessary with PgBouncer transaction mode. Without PgBouncer, or with session mode, leave the default (100).

How to distinguish: check the DATABASE_URL endpoint. If it points at PostgreSQL directly (port 5432) or at PgBouncer in session mode, you don't need the fix.

How to fix it: apply the fix only when it's appropriate. Have two configs per environment if your local doesn't use PgBouncer but production does (or, better, use PgBouncer locally too with docker-compose to avoid discrepancies).

Mistake 4 (operational): SET (without LOCAL) in transaction mode

Symptom: "My multitenant app works locally. In production, queries sometimes read data from the wrong tenant."

Why it happens: SET (without LOCAL) only affects the current connection. In transaction mode, the next query may use another real connection where SET doesn't apply.

How to distinguish: queries return data from other tenants occasionally. Non-deterministic, depends on how PgBouncer assigns connections.

How to fix it: use SET LOCAL inside explicit transactions (async with session.begin():). A mandatory refactor.

Mistake 5 (operational): LISTEN/NOTIFY listeners connected to PgBouncer

Symptom: "My realtime notification system worked before migrating to PgBouncer. Now I don't receive events."

Why it happens: transaction mode breaks LISTEN/NOTIFY (the real connection changes between transactions).

How to distinguish: the listeners don't receive notifications, even though another session does NOTIFY correctly.

How to fix it: connect the listeners directly to PostgreSQL (port 5432, without PgBouncer). Use two engines: one for HTTP requests via PgBouncer, another for listeners directly to PG. Example in gotcha 3 above.

Exercises

Exercise 1: reproduce the bug

Configure PgBouncer in transaction mode (POOL_MODE: transaction). Launch an asyncpg script with the cache enabled and many concurrent workers. Reproduce the InvalidSQLStatementNameError.

# docker-compose.yml
pgbouncer:
  environment:
    POOL_MODE: transaction
    DEFAULT_POOL_SIZE: "5"  # low, forces recycles
    MAX_CLIENT_CONN: "50"

docker compose restart pgbouncer

# reproduce_bug.py
import asyncio
import asyncpg

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

async def worker(pool, idx):
    fail_count = 0
    for i in range(100):
        try:
            async with pool.acquire() as conn:
                await conn.fetch("SELECT * FROM books WHERE id = $1", (i % 100) + 1)
        except asyncpg.exceptions.InvalidSQLStatementNameError as e:
            fail_count += 1
    return idx, fail_count


async def main():
    # NOTE: WITHOUT statement_cache_size=0 → cache enabled by default
    pool = await asyncpg.create_pool(DSN, min_size=10, max_size=20)

    results = await asyncio.gather(*[worker(pool, i) for i in range(30)])

    total_fails = sum(f for _, f in results)
    print(f"Total failures: {total_fails} of {30 * 100} requests")
    for idx, f in results:
        if f > 0:
            print(f"  Worker {idx}: {f} failures")

    await pool.close()


asyncio.run(main())

Total failures: 47 of 3000 requests
  Worker 3: 5 failures
  Worker 7: 8 failures
  Worker 12: 3 failures
  ...

# Change the pool creation to:
pool = await asyncpg.create_pool(
    DSN,
    min_size=10,
    max_size=20,
    statement_cache_size=0,  # ← FIX
)

Total failures: 0 of 3000 requests

Exercise 2: apply the fix in SQLAlchemy + verify

Take your app/db.py (from capsule 04) and apply the fix. Verify that it still works with PgBouncer in transaction mode.

engine = create_async_engine(
    "postgresql+asyncpg://bookstore:bookstore@localhost:6432/bookstore",
    pool_size=20,
    max_overflow=0,
    pool_timeout=10,
    pool_pre_ping=True,
    pool_recycle=3600,
    connect_args={
        "statement_cache_size": 0,
        "prepared_statement_cache_size": 0,
        "server_settings": {
            "application_name": "bookstore-api",
        },
    },
)

# In docker-compose.yml: POOL_MODE: transaction
docker compose restart pgbouncer

uvicorn app.main:app --reload

# Install wrk if you don't have it
brew install wrk  # macOS

# 30 seconds at 50 connections
wrk -t4 -c50 -d30s http://localhost:8000/books/1

Running 30s test @ http://localhost:8000/books/1
  4 threads and 50 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency    18.45ms    5.20ms   89.30ms   80.12%
    Req/Sec     645.30    102.45    789.00     78.45%
  77432 requests in 30.10s, 18.95MB read
Requests/sec:   2572.49
Transfer/sec:    645.21KB

# If your app logs SQL errors, there should be no InvalidSQLStatementNameError
docker compose logs app | grep -i "InvalidSQLStatementName"
# No output = fine

Exercise 3: measure the real cost of the fix

Compare p50/p95 latency with and without the fix (both against PostgreSQL directly, without PgBouncer, to isolate the cache effect).

# bench_cache_real.py
import asyncio
import time
import statistics
import asyncpg

DSN = "postgresql://bookstore:bookstore@localhost:5432/bookstore"  # PG directly
N = 1000


async def benchmark(cache_size: int):
    conn = await asyncpg.connect(DSN, statement_cache_size=cache_size)

    # warmup
    for _ in range(50):
        await conn.fetch("SELECT * FROM books WHERE id = $1", 1)

    # measure individually
    latencies_us = []
    for i in range(N):
        start = time.perf_counter()
        await conn.fetch("SELECT * FROM books WHERE id = $1", (i % 100) + 1)
        elapsed_us = (time.perf_counter() - start) * 1_000_000
        latencies_us.append(elapsed_us)

    await conn.close()

    p50 = statistics.median(latencies_us)
    p95 = sorted(latencies_us)[int(N * 0.95)]
    p99 = sorted(latencies_us)[int(N * 0.99)]
    avg = statistics.mean(latencies_us)

    return {"avg": avg, "p50": p50, "p95": p95, "p99": p99}


async def main():
    print("With cache (default 100):")
    cached = await benchmark(100)
    print(f"  avg: {cached['avg']:.0f}us | p50: {cached['p50']:.0f}us | p95: {cached['p95']:.0f}us | p99: {cached['p99']:.0f}us")

    print("\nWithout cache (0):")
    no_cache = await benchmark(0)
    print(f"  avg: {no_cache['avg']:.0f}us | p50: {no_cache['p50']:.0f}us | p95: {no_cache['p95']:.0f}us | p99: {no_cache['p99']:.0f}us")

    print(f"\nOverhead p50: {(no_cache['p50']-cached['p50'])/cached['p50']*100:.1f}%")
    print(f"Overhead p95: {(no_cache['p95']-cached['p95'])/cached['p95']*100:.1f}%")


asyncio.run(main())

With cache (default 100):
  avg: 285us | p50: 270us | p95: 380us | p99: 520us

Without cache (0):
  avg: 345us | p50: 320us | p95: 450us | p99: 620us

Overhead p50: 18.5%
Overhead p95: 18.4%

Exercise 4: refactor SET search_path for transaction mode

Your multitenant app has this pattern (which breaks in transaction mode):

async def get_user_for_tenant(db: AsyncSession, tenant_id: str, user_id: int):
    await db.execute(text(f"SET search_path TO tenant_{tenant_id}"))
    return await db.scalar(select(User).where(User.id == user_id))

Refactor it so it works in PgBouncer transaction mode.

async def get_user_for_tenant(db: AsyncSession, tenant_id: str, user_id: int):
    async with db.begin():  # explicit transaction
        await db.execute(text(f"SET LOCAL search_path TO tenant_{tenant_id}"))
        return await db.scalar(select(User).where(User.id == user_id))
    # COMMIT on exit; the SET LOCAL ends with the transaction.

import pytest

@pytest.mark.asyncio
async def test_search_path_works_in_transaction():
    async with SessionLocal() as session:
        async with session.begin():
            await session.execute(text("SET LOCAL search_path TO tenant_42"))

            # Verify that we're in the correct schema
            result = await session.scalar(text("SHOW search_path"))
            assert "tenant_42" in result

            # Run a query that depends on the search_path
            user = await session.scalar(select(User).where(User.id == 1))
            assert user is not None  # should read from tenant_42

# Better: parameterize (although SET doesn't accept parameters directly, validate the input)
import re
if not re.match(r"^[a-z0-9_]+$", tenant_id):
    raise ValueError(f"Invalid tenant_id: {tenant_id}")
await db.execute(text(f"SET LOCAL search_path TO tenant_{tenant_id}"))

Exercise 5: strategy for LISTEN/NOTIFY with PgBouncer transaction mode

Your app has a notification system that uses LISTEN/NOTIFY. Design the architecture so it coexists with PgBouncer transaction mode for the rest of the app.

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

# Main engine: via PgBouncer transaction mode (all the HTTP traffic)
engine_main = create_async_engine(
    "postgresql+asyncpg://bookstore:bookstore@pgbouncer:6432/bookstore",
    pool_size=20,
    max_overflow=0,
    connect_args={
        "statement_cache_size": 0,
        "server_settings": {"application_name": "bookstore-api"},
    },
)

SessionMain = async_sessionmaker(engine_main, expire_on_commit=False)


# Engine for listeners: directly to PostgreSQL (without PgBouncer)
engine_listen = create_async_engine(
    "postgresql+asyncpg://bookstore:bookstore@postgres:5432/bookstore",  # ← port 5432, not 6432
    pool_size=2,           # few listeners
    max_overflow=0,
    pool_pre_ping=True,
    pool_recycle=3600,
    connect_args={
        "server_settings": {"application_name": "bookstore-listeners"},
    },
)

# app/notifications.py
import asyncio
import asyncpg
from sqlalchemy import text
from app.db import engine_listen

async def listen_for_orders():
    """Long-running task that listens for NOTIFY 'new_order' and processes it."""
    # We need a raw asyncpg connection (SQLAlchemy AsyncSession doesn't expose listeners cleanly)
    async with engine_listen.connect() as conn:
        raw_conn = await conn.get_raw_connection()
        asyncpg_conn = raw_conn.driver_connection

        await asyncpg_conn.add_listener("new_order", on_new_order)

        # Keep the connection alive
        while True:
            await asyncio.sleep(60)
            # Optional: ping to detect deaths
            try:
                await asyncpg_conn.execute("SELECT 1")
            except Exception:
                break

def on_new_order(connection, pid, channel, payload):
    print(f"New order received: {payload}")
    # Process the event async (enqueue in a queue, etc.)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: launch the listener in the background
    listener_task = asyncio.create_task(listen_for_orders())
    yield
    # Shutdown: cancel the listener, close the engines
    listener_task.cancel()
    await engine_main.dispose()
    await engine_listen.dispose()

@router.post("/orders")
async def create_order(
    order: OrderCreate,
    db: AsyncSession = Depends(get_db),  # ← uses engine_main (PgBouncer)
):
    new_order = Order(...)
    db.add(new_order)
    await db.commit()

    # Notify via the same session (PgBouncer)
    await db.execute(text("NOTIFY new_order, :payload"), {"payload": str(new_order.id)})
    await db.commit()
    # NOTIFY works from transaction mode (it's a simple atomic operation)
    # What breaks is LISTEN, not NOTIFY

    return new_order

                    ┌────────────────────────────┐
                    │  FastAPI                    │
                    │                             │
HTTP requests ──→   │  endpoints (Depends get_db) │
                    │  → engine_main              │ ──→ PgBouncer ──→ PostgreSQL
                    │                             │      (port 6432)    (port 5432)
                    │  background task            │
                    │  (listen_for_orders)        │
                    │  → engine_listen            │ ──→ PostgreSQL direct
                    │                             │      (port 5432)
                    └────────────────────────────┘

Summary and next step

In this capsule you:

• Understood why asyncpg + PgBouncer transaction mode break prepared statements.
• Applied the fix: statement_cache_size=0 in connect_args.
• Measured the real cost of the fix (~20% relative, <0.1ms absolute on simple queries).
• Reviewed other transaction-mode gotchas: SET LOCAL, LISTEN/NOTIFY, session advisory locks.
• Designed strategies to coexist with features that break (separate engines, refactor to SET LOCAL in a transaction).
• Have the final app/db.py configuration ready for production.

Before moving on, you should be able to:

• Recognize the InvalidSQLStatementNameError error and know that the fix is statement_cache_size=0.
• Apply the fix in any FastAPI + asyncpg app.
• Refactor SET search_path to SET LOCAL inside a transaction.
• Design an architecture with separate engines when LISTEN/NOTIFY is necessary.

Next capsule — pool sizing formulas and monitoring. You have the PgBouncer + asyncpg setup working correctly. Now the operational question: how many connections are the right ones? You're going to learn the classic HikariCP formula (((cores × 2) + spindles)), why it doesn't apply directly to async, how to measure under real load, and how to monitor continuously with SHOW POOLS + pg_stat_database + exported metrics. It's the capsule that closes the loop: correct setup + correct sizing + continuous observability = a production pool.

Resources

1. asyncpg — statement_cache_size — the official reference for the parameter.
2. SQLAlchemy 2.0 — asyncpg dialect — the specific section on the prepared statement cache.
3. PgBouncer — features matrix — what breaks in each mode, official.
4. Supabase — "Choosing a connection pooling mode" — a practical discussion of transaction vs session.
5. GitHub issue: asyncpg + PgBouncer prepared statements — the thread where the bug and the fix were discussed.
6. Crunchy Data — PgBouncer prepared statements — a deep technical analysis.
7. PostgreSQL docs — Advisory locks — the difference between pg_advisory_lock and pg_advisory_xact_lock.

Module 6 — Database Performance & Query Tuning Guide