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

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)

如果对象是协程函数（使用 async def 语法定义的函数），则返回 true。

此函数从 Python 3.5 开始可用。 该模块可用于功能较少的 Python 2，当然没有您正在寻找的模块：inspect

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

inspect 模块提供了几个有用的函数来帮助获取有关活动对象的信息，例如模块、类、方法、函数、回溯、框架对象和代码对象。例如，它可以帮助您检查类的内容、检索方法的源代码、提取和格式化函数的参数列表，或者获取显示详细回溯所需的所有信息。

该模块主要提供四种服务：类型检查、获取源代码、检查类和函数以及检查解释器堆栈。

这个模块的一些基本功能是：

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

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

import asyncio

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

Answer 3

TLDR

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

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

---

详细

当您传递协程函数时，上述解决方案适用于简单的情况。在某些情况下，您可能希望传递 awaitable object 函数，该函数的作用类似于协程函数，但不是协程函数。两个例子是Future 类或Future-like object 类（implements__await__ 魔术方法的类）。在这种情况下，iscoroutinefunction 将返回 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 4

协程设置了 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 就是这样做的；测试对象是否是函数或方法（以确保存在__code__ 属性）并测试该标志。请参阅source code。

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

>>> inspect.iscoroutinefunction(foo)
True

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