Turn a sync function into an awaitable function

user_2963820 0
Viewed 181

I've got a situation where I can't be sure if a function will be sync, or async. In other words, the function has a type signature of Union[Callable[..., Awaitable[Any]], Callable[..., Any]].

I can't seem to find a good (non-deprecated) way of turning the above type signature into a consistent type of Callable[..., Awaitable[Any]].

The only way I was able to find how to do this, is by doing this:

import asyncio
import inspect
from typing import Any, Awaitable, Callable, List, Union

def force_awaitable(function: Union[Callable[..., Awaitable[Any]], Callable[..., Any]]) -> Callable[..., Awaitable[Any]]:
    if inspect.isawaitable(function):
        # Already awaitable
        return function
    else:
        # Make it awaitable
        return asyncio.coroutine(function)

However, asyncio.coroutine (which is normally used as a decorator) is deprecated since Python 3.8. https://docs.python.org/3/library/asyncio-task.html#asyncio.coroutine

The alternative provided does not work for me here, since I don't use asyncio.coroutine as a decorator. Unlike a decorator, the async keyword can't be used as a function.

How do I turn a sync function (not awaitable) into an async function (awaitable)?

Other considered options

Above I've shown that you can detect if something is awaitable. This allows us to change how to call like so:

def my_function():
    pass

if inspect.isawaitable(function):
    await my_function()
else:
    my_function()

However, this feels clunky, can create messy code, and creates unnecessary checks inside big loops. I want to be able to define how to call the function before I'm entering my loop.

In NodeJS I would simply await the sync function:

// Note: Not Python!
function sync() {}
await sync();

When I try to do the same thing in Python, I'm greeted with an error:

def sync():
  pass
await sync() # AttributeError: 'method' object has no attribute '__await__'
async-await asynchronous python
1 Answers

You can return an async function with the original function object in its scope:

def to_coroutine(f:Callable[..., Any]):
   async def wrapper(*args, **kwargs):
      return f(*args, **kwargs)
   return wrapper

def force_awaitable(function: Union[Callable[..., Awaitable[Any]], Callable[..., Any]]) -> Callable[..., Awaitable[Any]]:
   if inspect.iscoroutinefunction(function):
      return function
   else:
      return to_coroutine(function)

Now, if function is not awaitable, force_awaitable will return a coroutine function that contains function.

def test_sync(*args, **kwargs):
   pass

async def main():
   await force_awaitable(test_sync)('foo', 'bar')

asyncio.run(main())
user_73267380
Related