如何使用 FastAPI websocket 使 Postgres LISTEN 异步（非阻塞）？

Question

我正在尝试使 Postgress LISTEN 与 FastAPI 异步，以便在等待 Postgres 表更新时 WebSocket 连接不会阻塞。

到目前为止我得到了什么：

router = APIRouter()

@router.websocket("/pg_notify")
async def get_notifications(websocket: WebSocket):

    await websocket.accept()

    conn = psycopg2.connect("*****")
    conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
    curs = conn.cursor()
    curs.execute("LISTEN channel;")

    while True:
        try:
            conn.poll()
            while conn.notifies:
                print("Waiting for notification...")
                notify = await conn.notifies.pop(0)
                print(notify.payload)
        except Exception as e:
            print("exception triggered: ", str(e))
            await websocket.close()

这样会在 conn.notifies,pop(0) 上引发异常：

object psycopg2.extensions.Notify can't be used in the 'await' expression

Answer 1

最后以下对我有用。

while True:
        try:
            async with aiopg.create_pool(dsn) as pool:
                async with pool.acquire() as conn:
                    listener = listen(conn, "channel")

                    task_pg_listen = asyncio.create_task(listener)
                    result = await task_pg_listen