Test if function or method is normal or asynchronous

Question

How can I find out if a function or method is a normal function or an async function? I would like my code to automatically support normal or async callbacks and need a way to test what type of function is passed.

async def exampleAsyncCb():
    pass

def exampleNomralCb():
    pass

def isAsync(someFunc):
    #do cool dynamic python stuff on the function
    return True/False

async def callCallback(cb, arg):
    if isAsync(cb):
        await cb(arg)
    else:
        cb(arg)

And depending on what type of function gets passed it should either run it normally or with await. I tried various things but have no idea how to implement isAsync().

Answer 1

If you don't want to introduce another import with inspect, iscoroutine is also available inside asyncio.

import asyncio

def isAsync(someFunc):
    return asyncio.iscoroutinefunction(someFunc)

Answer 2

TLDR

If you want to check something should be used with await, use inspect.isawaitable (as when you test something being callable() other than just being a function).

Unlike iscoroutine or iscoroutinefunction it will also work for Futures and objects that implement __await__ method.

---

Detailed

Solutions above will work for simple cases, when you pass coroutine function. In some cases you may like to pass awaitable object function that acts like coroutine function, but is not coroutine function. Two examples is Future class or Future-like object class (class that implements __await__ magic method). In this cases iscoroutinefunction will return False, what is not you need.

It's easier to understand on non-async example with passing non-function callable as callback:

class SmartCallback:
    def __init__(self):
        print('SmartCallback is not function, but can be used as function')

callCallback(SmartCallback)  # Should work, right?

Back to async world, a similar situation:

class AsyncSmartCallback:
    def __await__(self):
        return self._coro().__await__()

    async def _coro(self):
        print('AsyncSmartCallback is not coroutine function, but can be used as coroutine function')
        await asyncio.sleep(1)

await callCallback(AsyncSmartCallback)  # Should work, but oops! iscoroutinefunction(AsyncSmartCallback) == False

Way to solve it not to use iscoroutine or iscoroutinefunction, but use inspect.isawaitable instead. It works with ready object so you must create it first. In other words, solution I would advise to use:

async def callCallback(cb, arg):
    if callable(cb):
        res = cb()  # here's result of regular func or awaitable
        if inspect.isawaitable(res):
            res = await res  # await if awaitable
        return res  # return final result
    else:
        raise ValueError('cb is not callable')

It's more universal (and I'm sure logically correct) solution.

Answer 3

Co-routines have the COROUTINE flag set, bit 7 in the code flags:

>>> async def foo(): pass
>>> foo.__code__.co_flags & (1 << 7)
128   # not 0, so the flag is set.

The value 128 is stored as a constant in the inspect module:

>>> import inspect
>>> inspect.CO_COROUTINE
128
>>> foo.__code__.co_flags & inspect.CO_COROUTINE
128

The inspect.iscoroutinefunction() function does just that; test if the object is a function or method (to ensure there is a __code__ attribute) and test for that flag. See the source code.

Of course, using inspect.iscoroutinefunction() is the most readable and guaranteed to continue to work if ever the code flags were to change:

>>> inspect.iscoroutinefunction(foo)
True

Answer 4

Extend the answers above. There has been 4 types of functions since python 3.6 :

• function
• generator function
• coroutine function
• asynchronous generator function

If your application does not have prior knowledge about the type of the given function, it could be one of them above, the asynchronous function could be either coroutine function or asynchronous generator function . asyncio.iscoroutinefunction(someFunc) only checks whether a function is coroutine function, for asynchronous generator, you can use inspect.isasyncgenfunction(). The sample code is shown below :

import inspect, asyncio

def isAsync(someFunc):
    is_async_gen = inspect.isasyncgenfunction(someFunc)
    is_coro_fn = asyncio.iscoroutinefunction(someFunc)
    return is_async_gen or is_coro_fn

Answer 5

use asyncio.iscoroutine() for judging coroutine, and asyncio.isfuture()for judging task or future

import asyncio


async def task():
    await asyncio.sleep(0.01)
    print(1)

async def main():
    t = task()
    print(type(t))# <class 'coroutine'>
    print(asyncio.iscoroutine(t)) # True
    print(asyncio.isfuture(t)) # False
    await t

async def main2():
    t = asyncio.create_task(task())
    print(type(t)) # <class '_asyncio.Task'>
    print(asyncio.iscoroutine(t)) # False
    print(asyncio.isfuture(t)) # True
    await t

if __name__ == '__main__':
    asyncio.run(main())
    asyncio.run(main2())

Answer 6

What about applying EAFP here:

try:
    result = await cb()
except TypeError as err:
    if "can't be used in 'await' expression" in str(err):
        result = cb()
    else:
        raise

this also solves the problem, when cb is an instance of partial as well