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

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

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

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

或

如何测试包裹在 partial object 中的函数？

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

Answer 1

使用 Python 版本 < 3.8 你不能让 partial() 对象通过该测试，因为测试要求有一个 __code__ 对象直接附加到传递给 inspect.iscoroutinefunction() 的对象。

您应该改为测试 partial 包装的函数对象，可通过 partial.func attribute 访问:

>>> 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() 委托(delegate))中的相关代码为 updated to handle partial() objects ，您不再需要自己解包 partial() 对象。该实现使用相同的 while isinstance(..., functools.partial) 循环。