部分异步函数未被检测为异步

Question

我有一个函数，它接受常规和异步函数（不是协程，而是返回协程的函数）。

它在内部使用asyncio.iscoroutinefunction() test来查看它获得了哪种类型的功能。

最近，当我尝试创建部分异步函数时，它崩溃了。

在这个演示中，ptest 不被识别为一个协程函数，即使它返回一个协程，即ptest() 是一个协程。

import asyncio
import functools

async def test(arg): pass
print(asyncio.iscoroutinefunction(test))    # True

ptest = functools.partial(test, None)
print(asyncio.iscoroutinefunction(ptest))   # False!!

print(asyncio.iscoroutine(ptest()))         # True

问题原因很清楚，但解决方案却不是。

如何动态创建通过测试的部分异步函数？

或者

如何测试包裹在部分对象中的 func ？

任何一个答案都可以解决问题。

Answer 1

使用低于 3.8 的 Python 版本，您无法使partial()对象通过该测试，因为该测试要求将一个__code__对象直接附加到您传递给的对象inspect.iscoroutinefunction()。

您应该测试partial包装的函数对象，可通过partial.func属性访问：

>>> asyncio.iscoroutinefunction(ptest.func)
True

如果您还需要测试partial()对象，则针对以下对象进行测试functools.partial：

def iscoroutinefunction_or_partial(object):
    while isinstance(object, functools.partial):
        object = object.func
    return inspect.iscoroutinefunction(object)

在 Python 3.8（及更新版本）中，inspect模块（asyncio.iscoroutinefunction()委托给）中的相关代码已更新为处理partial()对象，您不再需要partial()自己解包对象。该实现使用相同的while isinstance(..., functools.partial)循环。

Answer 2

我通过替换所有实例来解决这个partial问题async_partial：

def async_partial(f, *args):
   async def f2(*args2):
       result = f(*args, *args2)
       if asyncio.iscoroutinefunction(f):
           result = await result
       return result

   return f2