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

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)

根据传递的函数类型,它应该正常运行或使用await运行.我尝试了各种各样的东西,但不知道如何实现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

协同例程具有COROUTINE标志集,代码标志中的位6:

>>> async def foo(): pass
>>> foo.__code__.co_flags & (2 << 6)
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()功能就是这样; 测试对象是否是函数或方法(以确保存在__code__属性)并测试该标志.查看源代码.

当然,inspect.iscoroutinefunction()如果代码标志要改变,使用是最可读的并保证继续工作:

>>> inspect.iscoroutinefunction(foo)
True

Answer 4

当你传递协程函数时,上面的解决方案适用于简单的情况.在某些情况下,您可能希望传递一个像协程函数一样的等待对象函数,但不是协程函数.两个例子是Future类或类似Future的对象类(实现 await魔术方法的类).在这种情况下inspect.isawaitable将返回callable(),你不需要什么.

通过将非函数callable作为回调传递,可以更容易理解非异步示例:

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

await 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,而是使用Future.它适用于就绪对象,因此您必须先创建它.换句话说,我建议使用的解决方案:

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

它更通用(我确信在逻辑上是正确的)解决方案.