[Question] Asynchronous NOTIFY listener

[tomkp75]

Hello,

How would one use an async function as the callback of conn.add_listener? I would like to relay specific notifications to API calls using aiohttp.ClientSession.

Thanks,

Tom

[elprans]

This should do the trick:

conn.add_listener(
    'CHANNEL', 
    functools.partial(asyncio.ensure_future, async_callback()),
)

[tomkp75]

Thanks. Would it be an idea for add_listener to take in an awaitable directly?

[charterchap]

I ended up with something like this:

async def listener(self):
    async def process(*args):
        print(args)
        # await self.q.put(args)

    if not self.conn_listen:
        self.conn_listen = await asyncpg.connect(dsn)

    await self.conn_listen.add_listener('my_notify_channel',
        lambda *args: loop.create_task(process(args))
    )

def main():
    listener_task=loop.create_task(listener())
    loop.run_forever()

[adriangabura]

I ended up with something like this:

async def listener(self):
    async def process(*args):
        print(args)
        # await self.q.put(args)

    if not self.conn_listen:
        self.conn_listen = await asyncpg.connect(dsn)

    await self.conn_listen.add_listener('my_notify_channel',
        lambda *args: loop.create_task(process(args))
    )

def main():
    listener_task=loop.create_task(listener())
    loop.run_forever()

Thank you! I needed this for FastAPI websockets. Your solution is very elegant!

[theo-brown]

I second this, it seems a bit weird that the asynchronous Postgres library can't deal with asynchronous callback functions!

[theo-brown]

Can you perform asynchronous database operations within the callback function?

My attempt to do it doesn't work:

def __init__(self):
    ...
    self.conn.add_listener('channel', self.on_notify)
    ...
 
def on_notify(self, *args):
        asyncio.create_task(self.async_on_notify(*args))

async def async_on_notify(self, *args):
       await self.conn.execute(...)

It appears to block at the self.conn.execute line

[elprans]

It appears to block at the self.conn.execute line

I'm not sure what you mean by "block" here. If you mean "coroutine pauses execution", then it's what await does by definition. If you wanted to create a background task in async_on_notify and do other things in that handler, then you have to use asyncio.create_task() or asyncio.gather or any other concurrency primitive in asyncio.

[fantix]

Can you perform asynchronous database operations within the callback function?

My attempt to do it doesn't work:

Yes you can, but you should avoid concurrently using the same connection from two Tasks, by using your own asyncio synchronization mechanism (asyncio.Lock for example, but not recommended because it needs to be used with extra caution) or simply borrowing another connection in the callback task.