Convert callback-style asynchronous API to async / await in Python

Overview

Starting with Python 3.5, you can use async / await for asynchronous programming. However, many existing libraries have a traditional callback-type asynchronous API, and async / await cannot be applied as it is. Therefore, we will convert such a callback type API into a form that can be used with async / await.

Callback Asynchronous API Example

Add arguments a and b on another thread and return the result in a callback Consider the async_add method below.

import time
import threading

def async_add(a, b, callback):
    def run():
        time.sleep(1)
        callback(a + b)
    thread = threading.Thread(target=run)
    thread.start()

Let's use this to do a 1 + 2 + 3 calculation. It will be as follows.

async_add(1, 2, lambda result1: \
        async_add(result1, 3, lambda result2: \
                print(result2)))

Callbacks are nested and complicated. I would like to wrap this so that I can use await to write an image like this:

result1 = await awaitable_async_add(1, 2)
result2 = await awaitable_async_add(result1, 3)
print(result2)

Implementation

code

import time
import threading
import asyncio

def async_add(a, b, callback):
    def run():
        time.sleep(1)
        callback(a + b)
    thread = threading.Thread(target=run)
    thread.start()

def awaitable_async_add(a, b, loop):
    f = asyncio.Future() # (1)
    def callback(result):
        loop.call_soon_threadsafe(
                lambda: f.set_result(result)) #(2)
    async_add(a, b, callback) # (1)
    return f # (1)

async def exec(loop):
    result1 = await awaitable_async_add(1, 2, loop)
    result2 = await awaitable_async_add(result1, 3, loop)
    print(result2)

loop = asyncio.get_event_loop() # (3)
loop.run_until_complete(exec(loop)) # (3)
loop.stop()

Execution result

6

Commentary

(1) When the awaitable_async_add method is called, it starts executing the async_add method and immediately returns an asyncio.Future object. At this point, the calculation process is not complete yet.

(2) The callback is called when the calculation process is completed. By calling the set_result () method of the asyncio.Future object with the processing result as an argument, the asyncio.Future object is notified of the completion of processing and the calculation result is returned.

(3) Acquire the event loop and execute the process until the asynchronous process is completed.

One thing to note here is loop.call_soon_threadsafe in (2). If you try to call f.set_result directly as shown below, an error will occur.

    def callback(result):
        f.set_result(result)

Execution result

RuntimeError: Non-thread-safe operation invoked on an event loop other than the current one

This error is due to the set_result method being called on a different thread than the event loop running in (3). Since asyncio.Future is not thread-safe, the set_result method must always be called in the same thread as the event loop. Therefore, by passing the process as a callback to the call_soon_threadsafe method of the event loop, set_result is called in the event loop.

reference

https://docs.python.org/3.6/library/asyncio.html