不会将部分异步函数检测为异步函数
问题描述
我有一个函数,它同时接受常规函数和异步函数(不是协同例程,而是返回协同例程的函数)。
在内部,它使用asyncio.iscoroutinefunction()test查看它获得了哪种类型的函数。
最近,它在我尝试创建部分异步函数时出现故障。
在此演示中,ptest未被识别为Couroute函数,即使它返回协程,即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中的函数?
任一答案都可以解决问题。
Python3.8版无法使partial()对象通过该测试,因为该测试要求有一个__code__对象直接附加到您传递给inspect.iscoroutinefunction()的对象。inspect.iscoroutinefunction()。__code____code____code__推荐答案使用Python3.8版无法使partial()对象通过该测试,因为该测试要求有一个对象直接附加到您传递给的对象
您应该测试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)
在Python3.8(和更高版本)中,inspect模块(asyncio.iscoroutinefunction()委托给的)中的相关代码是updated to handle partial() objects,您不再需要自己解开partial()对象。该实现使用相同的while isinstance(..., functools.partial)循环。
相关文章