Python type hinting for async function as function argument

Question

I am trying to make sure a function parameter is an async function. So I am playing around with the following code:

async def test(*args, **kwargs):
    pass

def consumer(function_: Optional[Coroutine[Any, Any, Any]]=None):
    func = function_

consumer(test)

But it doesn't work.

I am presented with the following error during type checking in pyCharm:

Expected type 'Optional[Coroutine]', got '(args: Tuple[Any, ...], kwargs: Dict[str, Any]) -> Coroutine[Any, Any, None]' instead

Can anyone give me some hints how to solve this ?

Solution 1

I can't help you too much, especially because right now (PyCharm 2018.2) this error is not raised in Pycharm anymore.

At present, type hints are somewhere between reliable metadata for reflection/introspection and glorified comments which accept anything the user puts in. For normal data structures this is great (my colleague even made a validation framework based on typing), but things get more complicated when callbacks and async functions come into play.

Take a look at these issues:

https://github.com/python/typing/issues/424 (open as of today) - async typing https://github.com/python/mypy/issues/3028 (open as of today) - var-args callable typing

I would go with:

from typing import Optional, Coroutine, Any, Callable


async def test(*args, **kwargs):
    return args, kwargs


def consumer(function_: Optional[Callable[..., Coroutine[Any, Any, Any]]] = None):
    func = function_
    return func


consumer(test)

I don't guarantee they meant exactly that, but my hint is built like this:

Optional - sure, can be None or something, in this case:

Callable - something which can be invoked with (), ... stands for any argument, and it produces:

Coroutine[Any, Any, Any] - this is copied from OP, and very general. You suggest that this function_ may be await-ed, but also receive stuff send()-ed by consumer, and be next()-ed / iterated by it. It may well be the case, but...

If it's just await-ed, then the last part could be:

Awaitable[Any], if you actually await for something or

Awaitable[None], if the callback doesn't return anything and you only expect to await it.

Note: your consumer isn't async. It will not really await your function_, but either yield from it, or do some loop.run_until_complete() or .create_task(), or .ensure_future().

Solution 2

You are looking for:

FuncType = Callable[[Any, Any], Coroutine[Any]]
def consumer(function_: FuncType = None):

Why is the type structured like that? If you declare a function async, what you actually do is wrap it in a new function with the given parameters, which returns a Coroutine.

---

Since this might be relevant to some people who come here, this is an example of an awaitable function type:

OnAction = Callable[[Foo, Bar], Awaitable[FooBar]]

It is a function that takes Foo, Bar and returns a FooBar