类型提示异步函数并传递给 asyncio.create_task

Question

在我的研究中，我发现对于输入异步函数的正确类型提示的普遍共识是Callable[..., Awaitable[Any]]。

在 Pycharm 中，我尝试这个并在传递给 asyncio.create_task 时遇到这个问题

import asyncio
from typing import Callable, Awaitable, Any

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

这是 Pycharm 的问题，还是我应该以另一种方式提示我的异步函数？

Answer 1

让我们分解一下：

1. 当我们声明一个返回一些值的异步函数时 - 我们基本上创建了一个函数，如本例所示（Python 3.11）：

   async def ccc():
       return "Hello World"
   
   type(ccc) # <--- function
   

   这个功能回报类型的对象协程

   async def ccc():
       return "Hello World"
   
   type(ccc()) # <--- coroutine, not str!
   

   那是因为这个异步函数需要等待为了真正返回我们最初希望它返回的东西——一个字符串。

   实际上，在异步函数中调用协程不会像docs 状态那样执行任何操作：

   如果一个对象可以在 await 表达式中使用，我们就说它是一个 awaitable 对象。许多 asyncio API 旨在接受可等待对象。

   这意味着您的 ccc 函数确实是可调用的，它获得一些“%d”个参数（因此省略号），它返回一个协程。

   所以至少可以说，类型提示是这样的：

   Callable[..., Coroutine]
   

   你可以更明确地使用 Coroutine，因为我不知道 ccc 返回什么。

   此外，我不确定你想如何调用fff，因为它不是异步的，你需要这样调用并等待asyncio.create_task：

   import asyncio
   from typing import Callable, Coroutine
   
   
   async def ccc():
       print("Hello")
       await asyncio.sleep(5)
       print("World")
       return 2
   
   
   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()
   

   笔记：我为自己的解释案例随机创建了ccc，因为您没有在问题本身中提供它:-)