python - 如何将第三方库中的函数转换为异步函数？

Question

我正在使用我的 Raspberry Pi 以及 pigpio 和 websockets 库。
我希望我的程序异步运行(即我将使用 async def main 作为入口点)。pigpio 库期望调用同步 callback function 以响应事件，这很好，但是在该回调中，我想从 websocket 库中调用另一个异步函数。
所以它看起来像:

def sync_cb(): # <- This can not be made async, therefore I can not use await
   [ws.send('test') for ws in connected_ws] # <- This is async and has to be awaited

目前我可以让它与:

def sync_cb():
    asyncio.run(asyncio.wait([ws.send('test') for ws in connected_ws]))

但是 docs 说不鼓励使用 asyncio.run。
所以我的同步回调需要调用 ws.send (也来自 third party library )，它与同步函数异步。
另一个有效的选择是:

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(asyncio.gather(*[ws.send(json.dumps(message)) for ws in connected_ws]))

但是创建和设置偶数循环的三行代码听起来很简单，只是为了运行一个简单的异步函数。
我的问题是:

是否可以替换需要同步回调的异步函数(即在此示例中是否有办法使 cb 异步)

而且，我使用 asyncio.run 和 asyncio.wait 来调用一个简单的异步方法(在列表理解中)会产生什么样的开销

Answer 1

您可以使用 run_coroutine_threadsafe 函数返回 concurrent.furures.Future ，可以同步等待，将协程包装到常规函数并从同步代码中调用它。
据我了解，如果同步代码(第三方库的)在单独的线程中执行，这种方法更合适，但它可以通过一些修改适应单线程执行。
举例说明该方法:

import asyncio


def async_to_sync(loop, foo):
    def foo_(*args, **kwargs):
        return asyncio.run_coroutine_threadsafe(foo(*args, **kwargs), loop).result()  
    return foo_


def sync_code(cb):
    for i in range(10):
        cb(i)


async def async_cb(a):
    print("async callback:", a)


async def main():
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, sync_code, async_to_sync(loop, async_cb))

asyncio.run(main())

输出:

async callback: 0
async callback: 1
async callback: 2
...