不会将部分异步函数检测为异步函数

2022-03-25 00:00:00 python python-asyncio

问题描述

我有一个函数,它同时接受常规函数和异步函数(不是协同例程,而是返回协同例程的函数)。

在内部,它使用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)循环。

相关文章