Equivalent of psycopg2's poll in psycopg>=3?

[jamesbraza]

I am trying to make a DB listener using psycopg>=3. For this, one would use Connection.poll with psycopg2.

How can polling be done with psycopg>=3?

I am trying to adapt this gist for psycopg v3

[dvarrazzo]

Please check the documentation here about how to receive notifications.

We already know that this area will require some improvement, in particular to make it easier to stop listening and to share the listening connection with other users, see #340. There is some work on the way in #673 to improve that.

[jamesbraza]

Thanks for the detailed response! Yeah I was looking at that the asynchronous notifications docs last night too, and I did figure out psycopg v3 removes poll() in favor of direct iteration over notifies(), which I think is an improvement 👍

Where I wasn't sure was, in that asynchronous notifications example, it requires another entity to actually push a notification. However, I just want to listen for DB inserts.

I made the below code to stub out a DB insert. When inserting a new item (2nd to last line), the notifies list remains empty, which means the notification never happened. Any ideas why the insert doesn't lead to a notification?

import asyncio
from threading import Thread

import psycopg
from src.database import engine

conn = psycopg.connect(
    host=engine.url.host,
    dbname=engine.url.database,
    user=engine.url.username,
    password=engine.url.password,
    autocommit=True,
)
conn.execute("LISTEN match_updates;")
notifies = []

def handle_notify() -> None:
    for notify in conn.notifies():
        notifies.append(notify)

def read_notifies() -> None:
    loop = asyncio.new_event_loop()
    loop.add_reader(conn, handle_notify)
    loop.run_forever()

thread = Thread(target=read_notifies)
thread.start()

db_item = FactoryBoyFactory()
_ = 0  # At this point, notifies list is still empty

[dvarrazzo]

I don't see why you are using asyncio and threads together. I also don't see where you are doing the insert and how.

[jamesbraza]

Thanks for looking into it. I was using asyncio for its add_reader to match this gist, and then Thread to move it to a thread so I can perform the DB insert in the main thread. The DB insert takes place via factory-boy in the second to last item db_item = FactoryBoyFactory()

Changing it to a formal insert (see below) doesn't lead to a notification either

with Session() as session, session.begin():
    session.add(instance=Product(name="foo", category=Categories.BAZ))

[dvarrazzo]

I don't see a query here. I don't see if there is a transaction, if the transaction is terminated.

Just connect to the same database using psql and run LISTEN match_updates 'hello': if it works, fine. We can't debug your layers of sqlalchemy and factoryboy: we are just concerned about the postgres connection layer here.

[jamesbraza]

Okay I will play around, and if I figure it out I will circle back. Thank you again for the pointers, I really appreciate 👍