Typehinting async function and passing to asyncio.create_task

Question

In my research, I see the general consensus for the correct way to typehint an async function is Callable[..., Awaitable[Any]].

In Pycharm, I try this and have this issue when passing to asyncio.create_task

import asyncio
from typing import Callable, Awaitable, Any

def fff(ccc: Callable[..., Awaitable[Any]]):
    return asyncio.create_task(ccc())

Is this an issue with Pycharm, or should I be typehinting my async functions another way?

Answer 1

Let's break things down to pieces:

1. When we declare an asynchronous function that returns some value - we basically create a function as this example shows (Python 3.11):

async def ccc():
    return "Hello World"

type(ccc) # <--- function

2. This function returns an object of type coroutine

async def ccc():
    return "Hello World"

type(ccc()) # <--- coroutine, not str!

That is because this asynchronous function needs to be awaited in order to really return what we've initially wanted it to return - a string.

Effectively, calling a coroutine inside an asynchronous function will not do anything as the docs state:

We say that an object is an awaitable object if it can be used in an await expression. Many asyncio APIs are designed to accept awaitables.

Meaning that your ccc function is indeed callable which gets some "%d" parameters (hence the ellipsis), which returns a coroutine.

So to say the least, the type hinting is something as such:

Callable[..., Coroutine]

And you can be more explicit with the Coroutine as I don't know what ccc returns.

Moreover, I am not sure how you want to call fff as it is not asynchronous, you will need to make it such and await the asyncio.create_task as such:

import asyncio
from typing import Callable, Coroutine


async def ccc():
    print("Hello")
    await asyncio.sleep(5)
    print("World")
    return 12345


async def fff(random_callable: Callable[..., Coroutine]):
    result = await asyncio.create_task(random_callable())
    print(f"{result=}")


if __name__ == '__main__':
    el = asyncio.get_event_loop()
    el.run_until_complete(fff(ccc))
    el.close()

Note: I created ccc randomly for my own case of explanation, as you did not provide it in the question itself :-)