如何避免已取消任务未执行协程的RuntimeWarning警告？

当你创建runIt()协程并传入delegate()，再将delegate封装为Task后立即取消，由于runIt从未被执行，会触发RuntimeWarning提示协程未被等待。

原代码如下：

import asyncio

async def cancelTaskTest():
    async def runIt():
        print("RunIt ran")

    async def delegate(coro):
        await coro

    task = asyncio.create_task(delegate(runIt()))
    task.cancel()

if __name__=='__main__':
    asyncio.run(cancelTaskTest())

执行后触发的警告：

/usr/lib/python3.10/asyncio/base_events.py:1881: RuntimeWarning: coroutine 'cancelTaskTest.<locals>.runIt' was never awaited
  handle = self._ready.popleft()
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

已知runIt未运行，如何消除该警告？

最佳解决方法 #

方法一：在delegate中处理取消，手动关闭未执行的协程 #

修改delegate函数，捕获取消异常并关闭未被等待的runIt协程，同时确保Task正常完成：

import asyncio

async def cancelTaskTest():
    async def runIt():
        print("RunIt ran")

    async def delegate(coro):
        try:
            await coro
        except asyncio.CancelledError:
            # 关闭未被执行的协程，消除警告
            coro.close()
            raise  # 重新抛出异常，保持Task的取消状态

    task = asyncio.create_task(delegate(runIt()))
    task.cancel()
    # 等待Task完成，避免asyncio退出时的额外警告
    try:
        await task
    except asyncio.CancelledError:
        pass

if __name__=='__main__':
    asyncio.run(cancelTaskTest())

方法二：延迟创建runIt协程（更简洁） #

将runIt函数而非已创建的协程传入delegate，仅在delegate执行时才创建协程。这样取消Task时，runIt协程还未被实例化，自然不会产生未等待的警告：

import asyncio

async def cancelTaskTest():
    async def runIt():
        print("RunIt ran")

    async def delegate(coro_func):
        await coro_func()  # 按需创建协程

    task = asyncio.create_task(delegate(runIt))  # 传入函数对象
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass

if __name__=='__main__':
    asyncio.run(cancelTaskTest())

内容的提问来源于stack exchange，提问作者user48956