How to convert a function in a third party library to be async?

Question

I am using my Raspberry Pi and the pigpio and websockets libraries.

I want my program to run asynchronously (i.e. I will use async def main as the entry point).

The pigpio library expects a synchronous callback function to be called in response to events, which is fine, but from within that callback I want to call another, asynchronous function from the websocket library.

So it would look like:

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

Currently I can get it to work with:

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

but the docs say this use of asyncio.run is discouraged.

So my synchronous callback needs to call ws.send (also from a third party library) which is async from a function that is synchronous.

Another option that works is:

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]))

But the three lines of creating and setting an even loop sounds like a lot just to run a simple async function.

My questions are:

• Is it possible to substitute an async function where a synchronous callback is required (i.e. is there a way to make cb async in this example)
• And, what kind of overhead am I incurring by using asyncio.run and asyncio.wait just to call a simple async method (in the list comprehension)

Answer 1

You could use run_coroutine_threadsafe function returning concurrent.furures.Future, which can be waited synchronously, to wrap coroutine to regular function and call it from synchronous code.

As I understand it, this approach is more appropriate if sync code (of third party lib) is executed in separate thread, but it can be adapted to single-threaded execution with some modifications.

An example to illustrate the approach:

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())

Output:

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