classic-pg-listener

Библиотека дает простой класс Listener для добавления реакций на уведомления, передаваемые PostgreSQL при использовании команд LISTEN и NOTIFY.

Для использования необходимо передать функцию-фабрику, порождающую соединение. Соединение должно быть с autocommit=True.

Для добавления коллбеков используется метод .add_callback(). В него следует передать коллбек и название канала. Коллбек будет вызван при приеме уведомления в соответствующем канале.

Методом .remove_callback() можно убрать коллбек, если он больше не нужен.

Например:

from classic.pg.listener import Listener
from psycopg import connect, Notify

listener = Listener(
lambda: connect(
user='test',
password='test',
host='127.0.0.1',
port='5432',
dbname='test',
autocommit=True,
)
)

def some_task(notify: Notify):
print(notify)

listener.add_callback(some_task, 'some_channel')
listener.run()

Может работать в двух режимах:

• "Блокирующий режим". При нем исполнение происходит в текущем потоке. Для этого используется блокирующий метод .run()

• "Фоновый режим". При нем запускается внутренний поток для обслуживания соединения с базой и приема сообщений. В этом режиме класс похож на класс Thread из модуля threading. Метод .start() запускает фоновый поток, в котором происходит установка и обслуживание соединения, метод .stop() используется для запроса остановки фонового потока, методом .join() можно дождаться остановки внутреннего потока.

Пример использования в фоновом режиме:

from classic.pg.listener import Listener
from psycopg import connect, Notify

listener = Listener(
lambda: connect(
user='test',
password='test',
host='127.0.0.1',
port='5432',
dbname='test',
autocommit=True,
)
)

def some_task(notify: Notify):
print(notify)

listener.add_callback(some_task, 'some_channel')
listener.start()

print('Основной поток все еще свободен!')

listener.stop()
listener.join() # Дожидается остановки

Коллбеки, назначенные каналам, по умолчанию выполняются во внутреннем ThreadPoolExecutor. Параметром max_workers в конструкторе можно повлиять на размер пула. При max_workers исполнение будет происходить без пула, в том же потоке, что и обслуживает соединение. Таким образом, при запуске через метод .run() и значении max_workers=0 все, и обслуживание соединения, и исполнение коллбека будет происходить в одном потоке.

Методы .add_callback() и .remove_callback() выполняют свою работу не сразу. Оба метода возвращают future, в результате которого будет None, если добавление произошло успешно, либо выкинет исключение, произошедшее в потоке, обслуживающем слушателя.

Listener умеет регулярно опрашивать БД, в случае, если обновления не приходят в определённый период. Период устанавливается параметром ping_period в конструкторе Listener. По умолчанию 1 минута. Если ping_period=0, то проверка выполняться не будет.