Partial of a class coroutine isn't a coroutine. Why?

Question

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???

I need to find a way to wrap a coroutine inside a class into partial so that the result is also a coroutine. How do I do that?

Answer 1

The why is the way the inspect module checks for this.

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)

If we look at the definition for _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)

We see that it first it tries to unwrap a bound method from and get its .func attribute(which contains the function object) and then after that, unwrap the partial. Finally if the result isn't a function return False otherwise return the result of a flag check on the underlying function's __code__ attribute.

The problem is that while ismethod(f) does nothing since at that point it is still a partial object. Then after it unwraps if from the partial, isfunction returns False because it is only a bound method there.

That is the why. I do not know if this could be considered a bug or if it was done so by design. The fact that the docstring for _has_code_flag leaves out functools.partial wrapped methods in its description leads me to believe it was by design.

However, you can borrow from functools._unwrap_partial and use their method of checking for a coroutine by checking the .func attribute.

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

Taken from this answer:

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

Answer 2

Based on your comment, you need to create a coroutine:

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

The difference between this and the "real" functools.partial is that the callable returned by the latter can be invoked multiple times.