测试函数或方法是正常的还是异步的

Question

如何判断函数或方法是普通函数还是异步函数？我希望我的代码自动支持正常或异步回调，并且需要一种方法来测试传递的函数类型。

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)

并且根据传递的函数类型，它应该运行通常或等待。我尝试了各种方法，但不知道如何实施 isAsync().

Answer 1

使用Python的inspect模块。

inspect.iscoroutinefunction(object)

Return true if the object is a coroutine function (a function defined with an async def syntax).

此功能自 Python 3.5 起可用。该模块可用于 Python 2，但功能较少，当然没有您要找的模块：inspect

顾名思义，Inspect 模块对检查很多东西很有用。文档说

The inspect module provides several useful functions to help get information about live objects such as modules, classes, methods, functions, tracebacks, frame objects, and code objects. For example, it can help you examine the contents of a class, retrieve the source code of a method, extract and format the argument list for a function, or get all the information you need to display a detailed traceback.

There are four main kinds of services provided by this module: type checking, getting source code, inspecting classes and functions, and examining the interpreter stack.

该模块的一些基本功能是：

inspect.ismodule(object)
inspect.isclass(object)
inspect.ismethod(object)
inspect.isfunction(object)

它还包含检索源代码的功能

inspect.getdoc(object)
inspect.getcomments(object)
inspect.getfile(object) 
inspect.getmodule(object)

方法命名直观。如果需要，可以在文档中找到说明。

Answer 2

协程设置了 COROUTINE 标志，代码标志中的第 7 位：

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

值 128 作为常量存储在 inspect 模块中：

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

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.

当然，使用 inspect.iscoroutinefunction() 是最易读的，并且保证在代码标志发生更改时继续工作：

>>> inspect.iscoroutinefunction(foo)
True

Answer 3

如果您不想使用 inspect 引入另一个导入，iscoroutine 也可以在 asyncio 中使用。

import asyncio

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

Answer 4

TLDR

如果你想检查某些东西应该与 await 一起使用，请使用 inspect.isawaitable（当你测试某些东西是 callable() 而不仅仅是一个函数时）。

与 iscoroutine 或 iscoroutinefunction 不同，它也适用于 Future 和实现 __await__ 方法的对象。

详细

当您传递协程函数时，上述解决方案适用于简单的情况。在某些情况下，您可能想传递 awaitable object 函数，它的行为类似于协程函数，但不是协程函数。两个例子是 Future class or Future-like object class (class that implements __await__ 魔术方法）。在这种情况下 iscoroutinefunction 将 return False，你不需要什么。

在非异步示例中将非函数调用作为回调传递更容易理解：

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

callCallback(SmartCallback)  # Should work, right?

回到异步世界，类似的情况：

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

解决方法不使用iscoroutine或iscoroutinefunction，而是使用inspect.isawaitable。它适用于现成的对象，因此您必须先创建它。换句话说，我建议使用的解决方案：

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')

它是更通用（而且我确信逻辑上正确）的解决方案。

Answer 5

扩展上面的答案。自 python 3.6 以来已有 4 types of functions :

函数
生成器函数
协程函数
异步生成器函数

如果您的应用程序没有关于给定函数类型的先验知识，它可能是上面的其中之一，异步函数可以是 协程函数 或 异步生成器函数 . asyncio.iscoroutinefunction(someFunc)只检查一个函数是否是协程函数，对于异步生成器，可以使用inspect.isasyncgenfunction()。示例代码如下所示：

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 6

在这里应用EAFP怎么样：

async def callCallback(cb, arg):
    try:
        await cb(arg)
    except TypeError:
        cb(arg)

这也解决了问题，当 cb 是 partial

的实例时

Answer 7

使用asyncio.iscoroutine()判断协程， asyncio.isfuture()用于判断任务或未来

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

测试函数或方法是正常的还是异步的

Test if function or method is normal or asynchronous

python

asynchronous

coroutine

python-3.x

python-3.5