Postgres SQLAlchemy Textbook Ch. 6 — LISTEN / NOTIFY for Pub/Sub

Chapter 6: LISTEN/NOTIFY. Postgres’s built-in pub/sub. Useful for low-latency wake-up, simple event delivery, and small-scale fanout.

NOTIFY basics

NOTIFY channel_name, 'payload';

LISTEN channel_name;

Subscriber connection receives notifications. Payload is up to 8000 bytes by default.

SQLAlchemy + asyncpg

asyncpg has native LISTEN/NOTIFY support; SQLAlchemy doesn’t add a high-level API. Drop to the driver:

from asyncpg import Connection

async def listen_for_events():
    conn: Connection = await asyncpg.connect(DATABASE_URL)
    await conn.add_listener("user_events", on_notify)
    while True:
        await asyncio.sleep(60)  # keep connection alive

async def on_notify(connection, pid, channel, payload):
    print(f"Got {channel}: {payload}")
    data = json.loads(payload)
    # process...

Run as a long-lived background task.

Notify from a transaction

async with session.begin():
    session.add(user)
    await session.execute(text("NOTIFY user_events, :payload"), {"payload": json.dumps({"id": 1})})

Notifies are delivered on commit, not on issue. If transaction rolls back: no notify sent.

Combine with outbox

For reliable delivery (notifies don’t have replay):

async with session.begin():
    session.add(user)
    session.add(OutboxEvent(type="user.created", payload={"id": user.id}))
    await session.execute(text("NOTIFY outbox"))

Worker:

async def outbox_worker():
    conn = await asyncpg.connect(DATABASE_URL)
    await conn.add_listener("outbox", lambda *a: asyncio.create_task(drain()))
    while True:
        await asyncio.sleep(60)

async def drain():
    async with AsyncSessionLocal() as session:
        events = await session.execute(
            select(OutboxEvent).where(OutboxEvent.published_at == None).limit(100).with_for_update(skip_locked=True)
        )
        for e in events.scalars():
            await bus.publish(e.type, e.payload)
            e.published_at = datetime.utcnow()
        await session.commit()

NOTIFY wakes the worker; SKIP LOCKED ensures parallel workers don’t double-publish.

Channels

Channels are strings; namespace as you like:

• events
• events:user
• events:user:42

Subscribers can listen to specific channels.

Payload limits

8000 bytes per notify. For larger: send an ID; consumer fetches details.

NOTIFY orders, '{"order_id": 12345}';

Consumer reads the order from the DB.

At-most-once

NOTIFY is best-effort:

• If subscriber’s connection drops and reconnects: missed notifies don’t replay.
• Pair with persistent log + polling for durability.

For at-least-once: outbox + polling on a slow tick + NOTIFY for fast wake-up.

Connection considerations

• LISTEN connections are session-scoped. Hold one connection per listener.
• Multiple LISTENs per connection are supported.
• Use a separate connection (not session pool) for listening.

Multi-tenant

await conn.execute(f"LISTEN tenant_{tenant_id}")

Per-tenant channels for fan-out. Watch out for too many channels; Postgres handles thousands fine.

Triggers

Auto-NOTIFY on row changes via trigger:

CREATE OR REPLACE FUNCTION notify_user_change() RETURNS trigger AS $$
BEGIN
    PERFORM pg_notify('user_changes', json_build_object('op', TG_OP, 'id', NEW.id)::text);
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER user_change_trigger
AFTER INSERT OR UPDATE OR DELETE ON users
FOR EACH ROW EXECUTE FUNCTION notify_user_change();

Application code doesn’t need to NOTIFY — DB does it. Subscribers get every change.

For event-driven UIs / cache invalidation / search index updates.

Compared to other systems

For sub-ms wake-up of background workers: LISTEN/NOTIFY is ideal.

For durable event streams: Kafka.

Common mistakes

1. Treating NOTIFY as durable

Connection blip → missed events. Use outbox for durability.

2. Big payloads

8KB limit. Use IDs.

3. Listening on session pool

Connection returns to pool; LISTEN lost. Dedicated connection.

4. Connection without keepalive

Long-idle connections die. Configure keepalives or run periodic SELECT 1.

5. Triggers for everything

Heavy write paths get slower. Use sparingly.

What’s next

Chapter 7: Partitioning.

Read this next

• Chapter 7 — Partitioning
• Postgres as a Queue 2026

Building something AI-, backend-, or data-heavy and want a second pair of eyes? I do consulting and freelance work — see my projects and ways to reach me at rajpoot.dev .