如何在 asyncio 引发 TimeoutError 之前由 asyncio 任务本身处理超时异常答案

【问题标题】：How to handle Timeout Exception by the asyncio task itself just before asyncio raise TimeoutError如何在 asyncio 引发 TimeoutError 之前由 asyncio 任务本身处理超时异常
【发布时间】：2022-12-18 04:00:36
【问题描述】：

由于一个用例，我的一个长时间运行的函数执行了多条指令。但我必须为其执行提供最长时间。如果函数不能在分配的时间内完成执行，它应该清理进度并返回。

让我们看一下下面的示例代码：

import asyncio

async def eternity():
    # Sleep for one hour
    try:
        await asyncio.sleep(3600)
        print('yay!, everything is done..')
    except Exception as e:
        print("I have to clean up lot of thing in case of Exception or not able to finish by the allocated time")


async def main():
    try:
        ref = await asyncio.wait_for(eternity(), timeout=5)
    except asyncio.exceptions.TimeoutError:
        print('timeout!')

asyncio.run(main())

“eternity”函数是长期运行的函数。问题是，如果出现异常或达到最大分配时间，该函数需要清理它造成的混乱。

附言“永恒”是一个独立的功能，只有它才能理解要清洁什么。

我正在寻找一种方法在超时之前在我的任务中引发异常，或者向任务发送一些中断或终止信号并处理它。
基本上，我想在 asyncio 引发 TimeoutError 并取得控制权之前在我的任务中执行一些代码。
另外，我正在使用 Python 3.9。
希望我能够解释这个问题。

【问题讨论】：

标签： python python-3.x python-asyncio coroutine

【解决方案1】：

您需要的是异步上下文管理器：

import asyncio

class MyClass(object):

    async def eternity(self):
        # Sleep for one hour
        try:
            await asyncio.sleep(3600)
            print('yay!, everything is done..')
        except Exception as e:
            print("I have to clean up lot of thing in case of Exception or not able to finish by the allocated time")

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc, tb):
        print("Do what you need here!")


async def main():
    try:
        async with MyClass() as my_class:
            ref = await asyncio.wait_for(my_class.eternity(), timeout=3)
    except asyncio.exceptions.TimeoutError:
        print('timeout!')


asyncio.run(main())

这是输出：

Do what you need here!
timeout!

有关详细信息，请查看here。

【讨论】：