类协程的一部分不是协程。为什么？答案

【问题标题】：Partial of a class coroutine isn't a coroutine. Why?类协程的一部分不是协程。为什么？
【发布时间】：2021-04-12 18:44:22
【问题描述】：

class Foo:
    async def foo(self, a):
        return a

async def bar(b):
    return b

asyncio.iscoroutinefunction(functools.partial(bar, 1)) # returns True, OK
asyncio.iscoroutinefunction(functools.partial(Foo().foo, 1)) # returns False, WHY???

我需要找到一种方法将类中的协程包装成部分，以便结果也是一个协程。我该怎么做？

【问题讨论】：

请注意，您的“类协程”没有classmethod 装饰器，因此它是一个应该接受self 的普通方法。你的实际用例是什么？
我想将这样的类协程传递给 aiohttp 服务器，但得到以下结果：aiohttp/aiohttp/worker.py"，第 72 行，在 _run raise RuntimeError( RuntimeError: wsgi app 应该是 Application或异步函数返回 Application，得到 functools.partial(>, )

标签： python python-3.x python-asyncio

【解决方案1】：

为什么inspectmodule会检查这个。

def iscoroutinefunction(obj):
    """Return true if the object is a coroutine function.
    Coroutine functions are defined with "async def" syntax.
    """
    return _has_code_flag(obj, CO_COROUTINE)

如果我们查看definition 中的_has_code_flag：

def _has_code_flag(f, flag):
    """Return true if ``f`` is a function (or a method or functools.partial
    wrapper wrapping a function) whose code object has the given ``flag``
    set in its flags."""
    while ismethod(f):
        f = f.__func__
    f = functools._unwrap_partial(f)
    if not isfunction(f):
        return False
    return bool(f.__code__.co_flags & flag)

我们看到它首先尝试解包绑定的方法并获取其.func 属性（其中包含函数对象），然后解包partial。最后，如果结果不是函数，则返回 False，否则返回对底层函数的 __code__ 属性进行标志检查的结果。

问题是while ismethod(f) 什么都不做，因为此时它仍然是partial 对象。然后如果从partial 解包后，isfunction 返回False，因为它只是那里的绑定方法。

这就是原因。我不知道这是否可以被认为是一个错误，或者它是否是设计使然。 _has_code_flag 的文档字符串在其描述中省略了 functools.partial 包装方法这一事实让我相信这是设计使然。

但是，您可以从functools._unwrap_partial 借用他们的方法，通过检查.func 属性来检查coroutine。

def _unwrap_partial(func):
    while isinstance(func, partial):
        func = func.func
    return func

取自this answer：

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

【讨论】：

【解决方案2】：

根据你的comment，你需要创建一个协程：

def async_partial(async_fn, *args):
    async def wrapped():
        return await async_fn(*args)
    return wrapped

foo1 = async_partial(Foo().foo, 1)
assert inspect.iscoroutinefunction(foo1)
assert asyncio.run(foo1()) == 1

this 和“真正的”functools.partial 的区别在于后者返回的 callable 可以被多次调用。

【讨论】：