Hi! How do I wrap asyncpg add_listener function in Trio? The add_listener function calls provided callback every time NOTIFY happens in Postgres:

await conn.add_listener(channel, lambda *a: print("callback:", a))

I would like to wrap for 'async for' consumption:

async for event in listen(channel): print("event":, event)

How do I do it?

(I've searched through docs, google, github issues and forum. But I couldn't find the solution.)

Hi! How do I wrap asyncpg add_listener function in Trio? The add_listener function calls provided callback every time NOTIFY happens in Postgres:

await conn.add_listener(channel, lambda *a: print("callback:", a))

I would like to wrap it for 'async for' consumption:

async for event in listen(channel): print("event":, event)

How do I do it?

(I've searched through docs, google, github issues and forum. But I couldn't find the solution.)

Hi! How do I wrap asyncpg add_listener function in Trio? The add_listener function calls provided callback every time NOTIFY happens in Postgres:

await conn.add_listener(channel, lambda *a: print("callback:", a))

I would like to wrap it for 'async for' consumption:

async for event in listen(channel): print("event":, event)

How do I do it?

(I've searched through docs, google, github issues and forum. But I couldn't find the solution.)

@shamrin Here is a utility function I am using in my code

@asynccontextmanager
async def listen(channel, conn):
    send_channel, receive_channel = trio.open_memory_channel(1)

    def _listen_callback(c, pid, chan, payload):
        send_channel.send_nowait(orjson.loads(payload))

    await conn.add_listener(channel, _listen_callback)
    yield receive_channel
    await conn.remove_listener(channel, _listen_callback)

I have written a trigger that returns each inserted or updated record as json. This is then unpacked and sent to a memory channel, so I can use it in async code.

async with listen('my_channel', conn) as receive_channel:
    async for message in receive_channel:
        print(message)

In general, probably want:

async with send_channel:
    yield receive_channel

Right? (Though this is fine as-is in cases where both the producer and the consumer get cancelled at the same time, which is probably pretty often.)

In general, probably want:

async with send_channel:
    yield receive_channel

So the channel gets closed, right? (Though this is fine as-is in cases where both the producer and the consumer get cancelled at the same time, which is probably pretty often.)