Test if function or method is normal or asynchronous

user_3449303 0
Viewed 29400

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

python-3.5 coroutine asynchronous python-3.x python
7 Answers

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
user_98531050

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

user_43510270

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