python-3.x 测试函数或方法是否正常或异步

cwtwac6a  于 2023-06-25  发布在  Python
关注(0)|答案(8)|浏览(179)

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

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

rks48beu

rks48beu1#

使用Python的inspect模块。
inspect.iscoroutinefunction(object)
如果对象是协程函数(用async def语法定义的函数),则返回true。
这个函数从Python 3.5开始就可用了。该模块可用于Python 2,但功能较少,并且肯定没有您正在寻找的功能:inspect
顾名思义,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)

方法是直观地命名的。如有需要,可在文档中找到说明。

jw5wzhpr

jw5wzhpr2#

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

import asyncio

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

z3yyvxxp3#

TLDR

如果你想检查某个东西应该和await一起使用,使用inspect.isawaitable(当你测试某个东西是callable()而不仅仅是一个函数时)。
iscoroutineiscoroutinefunction不同,它也适用于Future和实现__await__方法的对象。

详细信息

上面的解决方案将适用于简单的情况,当你传递协程函数时。在某些情况下,您可能希望传递 awaitable object 函数,它的行为类似于协程函数,但不是协程函数。两个例子是Future类或 Future-like object 类(实现__await__魔术方法的类)。在这种情况下,iscoroutinefunction将返回False,这是你不需要的。
在非异步的例子中,传递非函数的callable作为callback更容易理解:

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

callCallback(SmartCallback)  # Should work, right?

回到async世界,类似的情况:

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

解决方法不是使用iscoroutineiscoroutinefunction,而是使用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')

这是更普遍的(我确信逻辑上是正确的)解决方案。

v64noz0r

v64noz0r4#

协同例程设置了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()函数就是这样做的;测试对象是否是函数或方法(以确保存在__code__属性)并测试该标志。参见源代码。
当然,使用inspect.iscoroutinefunction()是可读性最好的,并且可以保证在代码标志发生变化时继续工作:

>>> inspect.iscoroutinefunction(foo)
True
lmyy7pcs

lmyy7pcs5#

扩展上面的答案。自Python 3.6以来,有4种类型的函数:

  • 功能
  • 生成函数
  • 协同程序函数
  • 异步生成函数

如果您的应用程序不知道给定函数的类型,它可能是上述函数之一,异步函数可能是协程函数异步生成器函数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
nwlqm0z1

nwlqm0z16#

我在这里找不到其他答案来满足这个要求。至少在py 3.10中:

class Foo:
    async def __call__():
        pass

foo = Foo()
asyncio.iscoroutinefunction(foo)  #  produces False.

相反,为了测试你可以做:

asyncio.iscoroutinefunction(foo) or asyncio.iscoroutinefunction(foo.__call__)

请注意:

async def bar():
    pass

asyncio.iscoroutinefunction(bar.__call__)  # produces False

在我自己的代码中,我曾经有很多:

if asyncio.iscoroutinefunction(foo):
    await foo()
else:
    foo()

一个更清晰的模式,你想处理任何一个可能是:

async def if_coro(result):
    if asyncio.iscoroutine(result): # or inspect.iscoroutine,... and so on
        return await result
    else:
        return result

result = await if_coro(async_func())  # result is as expected
result = await if_coro(sync_func())  # result is as expected

可能存在上述的许多不同变型。

oxcyiej7

oxcyiej77#

如何在这里应用EAFP:

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

这也解决了当cb也是partial的instance时的问题
已知限制:

  • 同步函数返回等待。在这种情况下,逻辑将失败
qc6wkl3g

qc6wkl3g8#

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

相关问题