How can I make a set of functions that can be used synchronously as well as asynchronously?

Question

Imagine I have a set of functions like this:

def func1():
    func2()

def func2():
    time.sleep(1)  # simulate I/O operation
    print('done')

I want these to be usable synchronously:

# this would take two seconds to complete
func1()
func1()

as well as asynchronously, for example like this:

# this would take 1 second to complete
future = asyncio.gather(func1.run_async(), func1.run_async())
loop = asyncio.get_event_loop()
loop.run_until_complete(future)

The problem is, of course, that func1 somehow has to propagate the "context" it's running in (synchronously vs. asynchronously) to func2.

I want to avoid writing an asynchronous variant of each of my functions because that would result in a lot of duplicate code:

def func1():
    func2()

def func2():
    time.sleep(1)  # simulate I/O operation
    print('done')

# duplicate code below...
async def func1_async():
    await func2_async()

async def func2_async():
    await asyncio.sleep(1)  # simulate I/O operation
    print('done')

Is there any way to do this without having to implement an asynchronous copy of all my functions?

Answer 1

Here's my "not-an-answer-answer," which I know that Stack Overflow loves...

Is there any way to do this without having to implement an asynchronous copy of all my functions?

I don't think that there is. Making a "blanket translator" to convert functions to native coroutines seems next-to-impossible. That's because making a synchronous function asynchronous is about more than throwing an async keyword in front of it and a couple of await statements within it. Keep in mind that anything that you await must be awaitable.

Your def func2(): time.sleep(1) illustrates that point. Synchronous functions will make blocking calls, such as time.sleep(); asynchronous (native coroutines) will await non-blocking coroutines. Making this function asynchronous, as you point out, requires not just using async def func(), but awaiting asyncio.sleep(). Now let's say instead of time.sleep(), you're calling a more complex, blocking function. You build some sort of fancy decorator that slaps a function attribute called run_async, which is a callable, onto the decorated function. But how does that decorator know how to "translate" the blocking calls within func2() into their coroutine equivalents, if those are even defined? I can't think of any magic that would be smart enough to convert all of the calls in a synchronous function to their awaitable counterparts.

In your comments, you mention that this is for HTTP requests. For a real-world example the differences in call signatures and APIs between the requests and aiohttp packages. In aiohttp, .text() is an instance method; in requests, .text is a property. How could you build something smart enough to know differences such as that?

I don't mean to be discouraging--but I think that using threading would be more realistic.

Answer 2

So I found a way to achieve this, but since this is literally the first time I've done anything with async I can't guarantee that this doesn't have any bugs or that it's not a terrible idea.

The concept is actually pretty simple: Define your functions like normal asynchronous functions using async def and await where necessary, and then add a wrapper around them that automatically awaits the function if no event loop is running. Proof of concept:

import asyncio
import functools
import time


class Hybrid:
    def __init__(self, func):
        self._func = func

        functools.update_wrapper(self, func)

    def __call__(self, *args, **kwargs):
        coro = self._func(*args, **kwargs)

        loop = asyncio.get_event_loop()

        if loop.is_running():
            # if the loop is running, we must've been called from a
            # coroutine - so we'll return a future
            return loop.create_task(coro)
        else:
            # if the loop isn't running, we must've been called synchronously,
            # so we'll start the loop and let it execute the coroutine
            return loop.run_until_complete(coro)

    def run_async(self, *args, **kwargs):
        return self._func(*args, **kwargs)


@Hybrid
async def func1():
    await func2()

@Hybrid
async def func2():
    await asyncio.sleep(0.1)


def twice_sync():
    func1()
    func1()

def twice_async():
    future = asyncio.gather(func1.run_async(), func1.run_async())
    loop = asyncio.get_event_loop()
    loop.run_until_complete(future)


for func in [twice_sync, twice_async]:
    start = time.time()
    func()
    end = time.time()
    print('{:>11}: {} sec'.format(func.__name__, end-start))

# output:
#  twice_sync: 0.20142340660095215 sec
# twice_async: 0.10088586807250977 sec

However, this approach does have its limitations. If you have a synchronous function calling a hybrid function, calling the synchronous function from an asynchronous function will change its behavior:

@hybrid
async def hybrid_function():
    return "Success!"

def sync_function():
    print('hybrid returned:', hybrid_function())

async def async_function():
    sync_function()

sync_function()  # this prints "Success!" as expected

loop = asyncio.get_event_loop()
loop.run_until_complete(async_function())  # but this prints a coroutine

Take care to account for this!