LISTEN and NOTIFY

PostgreSQL’s LISTEN and NOTIFY let one session announce something on a named channel and every session listening on that channel receive it, with no table to poll. The driver wraps them in four methods and an iterator.

Listening

Listen on a connection in autocommit. LISTEN is transactional like any statement, so on a connection that is not in autocommit it takes effect only when you commit, and a notification sent before then never arrives.

>>> listener = agensgraph.connect("host=localhost dbname=docs user=agens", autocommit=True)
>>> listener.listen("new movies")
>>> listener.listening()
['new movies']

A channel name is quoted like any identifier, so a name with a space or a capital letter is taken as written. unlisten("new movies") stops listening on one channel, and unlisten() with no name on all of them.

Announcing

>>> sender = agensgraph.connect("host=localhost dbname=docs user=agens", autocommit=True)
>>> sender.notify("new movies", "Dune")
>>> sender.notify("new movies", "Arrival")

notify(channel, payload="") sends through pg_notify(), with the channel as a parameter, so the same name reaches the same listeners however it is spelled. The payload is text. A notification sent inside a transaction is delivered when that transaction commits, and not at all if it rolls back.

Receiving: an iterator

>>> for note in listener.notifications(timeout=5, stop_after=2):
...     print(note.channel, note.payload)
new movies Dune
new movies Arrival

notifications() yields each notification as it arrives, as an agensgraph.Notify with channel, payload and pid, the process id of the sending session. timeout is how long the whole loop may last, in seconds, not how long it waits for each notification; stop_after stops after that many. With neither, it waits for ever. In awaiting code it is an asynchronous iterator: async for note in aconn.notifications(...).

Receiving: a handler

A handler is a function called with each notification:

>>> received = []
>>> def on_note(note):
...     received.append(note.payload)
>>> listener.add_notify_handler(on_note)
>>> sender.notify("new movies", "Tenet")
>>> listener.execute("SELECT 1")
<agensgraph.connection.Cursor ...>
>>> received
['Tenet']

A handler is psycopg’s, and psycopg calls it while the connection processes the server’s messages, that is, while it runs a statement or waits for one. A connection that only listens and never runs anything never gets the chance; give it something to do, or use the iterator.

Use one or the other, not both. A connection with a handler refuses notifications(), because with both in place each notification would go to whichever route happened to be looking:

>>> list(listener.notifications(timeout=0.2))
Traceback (most recent call last):
  ...
RuntimeError: this connection already has a notify handler, and reading announcements both ways at once delivers each one to whichever route happens to be looking. Remove the handler, or read them through it alone.

Removing the handler makes the iterator usable again:

>>> listener.remove_notify_handler(on_note)

Announcing changes to the graph

A label is a table, so a trigger on it can announce every vertex written to it, whoever writes it:

conn.execute("""
    CREATE FUNCTION moviekb.announce_movie() RETURNS trigger LANGUAGE plpgsql AS $$
    BEGIN
        PERFORM pg_notify('new movies', NEW.properties->>'title');
        RETURN NEW;
    END $$""")
conn.execute("""
    CREATE TRIGGER announce AFTER INSERT ON moviekb.movie
    FOR EACH ROW EXECUTE FUNCTION moviekb.announce_movie()""")
conn.commit()

>>> conn.execute_query("CREATE (:movie {title: 'Dune: Part Two', year: 2024})")
Result(records=[], ...)
>>> conn.commit()
>>> [note.payload for note in listener.notifications(timeout=5, stop_after=1)]
['Dune: Part Two']

NEW.properties is the new element’s property map, and NEW.id its identity. The same works for an edge label, and for UPDATE and DELETE, with OLD for the element as it was.

>>> listener.close(); sender.close()