24

I have a function which accepts both regular and asynchronous functions (not coroutines, but functions returning coroutines).

Internally it uses asyncio.iscoroutinefunction() test to see which type of function it got.

Recently it broke down when I attempted to create a partial async function.

In this demonstration, ptest is not recognized as a couroutine function, even if it returns a coroutine, i.e. ptest() is a coroutine.

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

The problem cause is clear, but the solution is not.

How to dynamically create a partial async func which passes the test?

OR

How to test the func wrapped inside a partial object?

Either answer would solve the problem.

2 Answers 2

21

Using Python versions < 3.8 you can't make a partial() object pass that test, because the test requires there to be a __code__ object attached directly to the object you pass to inspect.iscoroutinefunction().

You should instead test the function object that partial wraps, accessible via the partial.func attribute:

>>> asyncio.iscoroutinefunction(ptest.func)
True

If you also need to test for partial() objects, then test against functools.partial:

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

In Python 3.8 (and newer), the relevant code in the inspect module (that asyncio.iscoroutinefunction() delegates to) was updated to handle partial() objects, and you no longer have to unwrap partial() objects yourself. The implementation uses the same while isinstance(..., functools.partial) loop.

Sign up to request clarification or add additional context in comments.

8 Comments

Thank you. Is the .func attr documented somewhere?
@VPfB: yes, the functools.partial() documentation states it exists explicitly; the Roughly equivalent to code shows the attributes that are set.
The functools docs had some problems with links (in 2018), that's why I did not find the info: bugs.python.org/issue34748
@DurandA: what is your exact question? How to test a partialmethod object wraps a coroutine function? Then it depends on exactly what was wrapped and how you accessed the partialmethod object. If you accessed it through ClassObj.partialmethod_name or instance.partialmethod_name, and it wrapped a normal async def method, then descriptor binding takes place and a partial() object is returned. So in that case the same technique applies.
@DurandA: actually, that's not quite correct. I filed a new bug report as I discovered that this is a little more complicated. For instance.partialmethod_name, you can simply unwrap first (and you have to do this in Python 3.8) too). For ClassObj.partialmethod_name you also have to look for the _partialmethod attribute, follow that, and then follow the .func attribute.
|
7

I solved this by replacing all instances of partial with async_partial:

def async_partial(f, *args):
   async def f2(*args2):
       result = f(*args, *args2)
       if asyncio.iscoroutinefunction(f):
           result = await result
       return result

   return f2

Comments

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.