anna-money / asyncpg-listen Goto Github PK

When I (pip) install asyncpg-listen and try to run the example code in the README.md, I get error "AttributeError: __aenter__" at line "async with asyncpg.connect() as connection:".

I'm running the latest asyncpg, 0.25.0, which doesn't support connections as context managers:
MagicStack/asyncpg#583
so I am not sure how you got the example to work as written.

Many thanks for sharing and maintaining this valuable project. Is there a way to use the connect_func with a connection from a connection pool?

pgpool = await asyncpg.create_pool(dsn=myl_dsn)
async with pgpool.acquire() as connection: 

I am trying to implement this in a FastAPI/Websockets app, so that when a table update has been made, the changes (or a simple message) is pushed back to the listening client. Each client could potentially be listening to different channels, in this example they are all listening to "customer2" .

What I need is to pass the websocket argument to the notification handler, so that instead of just printing it, it sends back the payload to whichever client was listening. This is where I am a bit stuck, as my knowledge of asyncio is limited. Any help most appreciated.

`
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()

try:
    listener = asyncpg_listen.NotificationListener(asyncpg_listen.connect_func(dsn))
    listener_task = asyncio.create_task(
      listener.run(
        {"customer2": handle_notifications},
        policy=asyncpg_listen.ListenPolicy.LAST,
        notification_timeout=50
        )
    )
    # add it to the list so they can be closed if the server is shutdown
    all_listener_tasks.append(listener_task)

except WebSocketDisconnect:
    logger.info('Client has disconnected') # maybe more needs to be done to clean up ....
    return

async def handle_notifications(notification: asyncpg_listen.NotificationOrTimeout) -> None:
print(f"{notification} has been received, sent to websocket ?")
`

This library requires Python 3.9 or greater, yet asyncpg itself has a supports 3.6 and later.
What is the technical requirement driving this decision (it's not obvious at first glance)?