PostgreSQL: LISTEN/NOTIFY

LISTEN/NOTIFY в PostgreSQL передаёт сообщение из одного сеанса всем сеансам, которые слушают его канал. У клиента есть listen() и notify(); NotificationListener — подписка, которая сама переподключается.

Прослушивание: client.listen(channel, callback)

listen() есть у клиента любого драйвера (client = Connections.get(connection_alias)). Открывает отдельное, не входящее в пул соединение, выполняет LISTEN на канале channel и вызывает callback(connection, pid, channel, payload) — обычную синхронную функцию, не корутину — для каждого полученного на нём NOTIFY. Возвращённый объект принадлежит вам — он никогда не берётся из пула и не возвращается в него.

Однако два драйвера возвращают разные типы:

• Клиент rust_pg (hare.dialects.postgresql.drivers.rust_pg) — client.listen(...) возвращает pg.Listener. Его .is_closed() возвращает True и после явного await listener.close(), и если соединение оборвалось само (обрыв сети, завершённый процесс сервера и т. п.) — он не просто запоминает, вызывали ли вы .close(). Если отбросить последнюю ссылку, не вызвав .close(), отдельное соединение всё равно закроется, когда сборщик мусора Python удалит Listener, — это страховка от утечки соединения, а не замена явному .close() (момент сборки не гарантирован).
• Клиент asyncpg (hare.dialects.postgresql.drivers.asyncpg) — client.listen(...) возвращает обычный asyncpg.Connection с собственным .is_closed() из asyncpg. Отдельной обёртки Listener у этого драйвера нет. Параметры подключения, которые относятся только к пулу (min_size, max_size, max_queries, max_inactive_connection_lifetime, ...), к этому отдельному соединению не применяются; неизвестный параметр даёт ConfigurationError.

В любом случае закройте возвращённый объект, когда слушать больше не нужно:

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()

Отправка: client.notify(channel, payload="")

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

Выполняет SELECT pg_notify($1, $2) с параметрами — канал и содержимое передаются как значения, поэтому кавычки и экранирование не нужны. Есть и у клиента транзакции (async with Transactions.atomic() as connection: await connection.notify(...)): там PostgreSQL ставит уведомление в очередь и доставляет его только при фиксации транзакции — при откате оно не доставляется никогда. Вне транзакции уведомление доставляется сразу.

client.features.supports_listen_notify равно True на обоих драйверах PostgreSQL и False в SQLite и ClickHouse (там нет ни listen(), ни notify()).

Подписка, которая сама переподключается: NotificationListener

Соединение обычного client.listen(), которое оборвалось (завершённый процесс сервера, обрыв сети, отключение по простою), так и остаётся мёртвым. hare.contrib.notify.NotificationListener оборачивает его сторожем и переподключается:

from hare.contrib.notify import NotificationListener

async def on_book_returned(payload: str) -> None:  # обычный `def` тоже подойдёт
    ...

listener = NotificationListener(
    "default",                 # имя подключения
    "book_returned",           # канал
    on_book_returned,          # callback(payload: str), синхронный или асинхронный
    reconnect_attempts=None,   # None — пытаться бесконечно; N — сдаться после N неудачных попыток
    backoff=0.5,               # основа паузы backoff * 2**номер_попытки или явная последовательность пауз
    max_backoff_seconds=30.0,  # наибольшая одна пауза; None — без ограничения
)
await listener.start()
...
await listener.stop()

или async with NotificationListener(...) as listener: ....

• start() запускает фоновую задачу и возвращается, когда закончилась первая попытка LISTEN (подключились или не удалось и теперь повторяем в фоне). Повторный вызов ничего не меняет. Задача всегда использует общий клиент подключения, даже если start() вызван внутри транзакции.
• stop() отменяет эту задачу, закрывает отдельное соединение и отменяет ещё работающие асинхронные функции-обработчики, дожидаясь всего этого. Повторный вызов ничего не меняет.
• run() — сам цикл; дождитесь его через await, чтобы выполнять в своей задаче (stop() всё равно работает); возвращается, когда попытки переподключения исчерпаны.
• is_listening — открыто ли сейчас живое соединение LISTEN.
• Функция-обработчик получает только строку содержимого. Она выполняется в потоке цикла событий с переменными контекста (contextvars) кода, который начал слушать. Если она возвращает ожидаемый объект (async def), он выполняется как отслеживаемая фоновая задача. Исключения обоих видов записываются в журнал и никогда не передаются дальше — ошибка обработчика не обрывает подписку.
• Сторож и паузы: соединение проверяется раз в секунду; когда оно обрывается, слушатель переподключается. Подряд идущие неудачи (не удалось подключиться или соединение оборвалось, не пережив одной проверки) ждут backoff * 2**attempt секунд (или следующий элемент последовательности backoff, повторяя последний), но не больше max_backoff_seconds. После reconnect_attempts неудачных попыток слушатель пишет в журнал на уровне ERROR и останавливается. Счётчик обнуляется, когда соединение переживает проверку. Если слушатель уже слушал, переподключение, отклонённое как ошибка настройки (база удалена и создана заново), повторяется так же — на обоих драйверах.
• Без поддержки (SQLite, ClickHouse): start() и run() дают UnSupportedError до открытия соединения — у базы нет каналов уведомлений; опрашивайте таблицу.
• channel, который LISTEN не принимает, даёт ConfigurationError — на нём не начало бы слушать ни одно переподключение: пустой, не текст или с нулевым байтом — из конструктора, длиннее имён базы (63 байта в PostgreSQL) — из start() и run().

NOTIFY не сохраняется

Уведомление, отправленное, пока слушатель переподключается, теряется. Используйте LISTEN/NOTIFY, чтобы уменьшить задержку поверх надёжного источника данных (опроса таблицы), но никогда вместо него — именно так его использует OutboxRelay.