PostgreSQL: LISTEN/NOTIFY

PostgreSQL's LISTEN/NOTIFY passes a message from one session to every session listening on its channel. A client has listen() and notify(); NotificationListener is a subscription that reconnects by itself.

Listening: client.listen(channel, callback)

listen() is available on the client for either driver (client = Connections.get(connection_alias)). It opens a dedicated, non-pooled connection, LISTENs on channel, and invokes callback(connection, pid, channel, payload) — a plain synchronous callable, not a coroutine — for every NOTIFY received on it. You own the returned object — it's never drawn from or returned to the connection pool.

The two drivers return genuinely different types, though:

• rust_pg client (hare.dialects.postgresql.drivers.rust_pg) — client.listen(...) returns a pg.Listener. Its .is_closed() reports True both after an explicit await listener.close() and if the underlying connection dies on its own (a dropped network, a killed server process, ...) — it doesn't just track whether you called .close(). If you drop the last reference without calling .close(), the dedicated connection still closes once Python garbage-collects the Listener — a safety net against a leaked connection, not a substitute for calling .close() explicitly (collection timing isn't guaranteed).
• asyncpg client (hare.dialects.postgresql.drivers.asyncpg) — client.listen(...) returns a plain asyncpg.Connection, with asyncpg's own .is_closed(). There is no separate Listener wrapper on this driver. Pool-only connection parameters (min_size, max_size, max_queries, max_inactive_connection_lifetime, ...) are left out of that dedicated connection; an unknown parameter raises ConfigurationError.

Either way, you must close what's returned when you're done listening:

def on_notify(connection, pid, channel, payload):
    print(f"got {payload!r} on {channel}")

client = Connections.get("default")
listener = await client.listen("book_returned", on_notify)
...
await listener.close()

Sending: client.notify(channel, payload="")

await client.notify("book_returned", "42")

Runs a parameterized SELECT pg_notify($1, $2) — the channel and payload are bound values, so no quoting or escaping is needed. It is available on the transaction client too (async with Transactions.atomic() as connection: await connection.notify(...)); there PostgreSQL queues the notification and delivers it only when the transaction commits — on rollback it is never delivered. Outside a transaction it is delivered immediately.

client.features.supports_listen_notify is True on both PostgreSQL drivers and False on SQLite and ClickHouse (which have neither listen() nor notify()).

A self-reconnecting subscription: NotificationListener

A raw client.listen() connection that dies (killed server process, network drop, idle timeout) just stays dead. hare.contrib.notify.NotificationListener wraps it with a watchdog and reconnects:

from hare.contrib.notify import NotificationListener

async def on_book_returned(payload: str) -> None:  # a plain `def` works too
    ...

listener = NotificationListener(
    "default",                 # connection alias
    "book_returned",           # channel
    on_book_returned,          # callback(payload: str), sync or async
    reconnect_attempts=None,   # None = retry forever; N = give up after N failed retries
    backoff=0.5,               # base of backoff * 2**attempt, or an explicit sequence of delays
    max_backoff_seconds=30.0,  # cap for a single delay; None = uncapped
)
await listener.start()
...
await listener.stop()

or async with NotificationListener(...) as listener: ....

• start() starts a background task and returns once the first LISTEN attempt has finished (connected, or failed and now retrying in the background). Idempotent. The task always uses the alias's shared client, even when start() is called inside a transaction.
• stop() cancels that task, closes the dedicated connection and cancels still-running async callbacks, waiting for all of it. Idempotent.
• run() is the loop itself — await it directly to run it in a task you own (stop() still works); it returns once the reconnect budget is exhausted.
• is_listening — whether a live LISTEN connection is open right now.
• Callback receives only the payload string. It runs on the event loop's thread with the contextvars of the code that started listening. If it returns an awaitable (an async def), that runs as a tracked background task. Exceptions from either type are logged, never propagated — a failing callback doesn't kill the subscription.
• Watchdog/backoff: the connection is checked once a second; when it dies, the listener reconnects. Consecutive failures (a failed connect, or a connection that dies before surviving one check) wait backoff * 2**attempt seconds (or the next entry of a backoff sequence, the last one repeating), clamped to max_backoff_seconds. After reconnect_attempts failed retries it logs at ERROR and stops. The counter resets once a connection survives a health check. Once it listened, a reconnect refused as configuration (its database dropped and made again) is retried the same way — on both drivers.
• No support (SQLite, ClickHouse): start() and run() raise UnSupportedError before any connection is opened — the database has no notification channels; poll instead.
• A channel no LISTEN takes raises ConfigurationError, as no reconnect would ever listen on it: empty, not a text or holding a null byte from the constructor, longer than the database's names (63 bytes on PostgreSQL) from start() and run().

NOTIFY is not persistent

A notification sent while the listener is reconnecting is lost. Use LISTEN/NOTIFY to cut latency on top of a durable source of truth (polling a table), never instead of it — exactly how OutboxRelay uses it.