为asyncio程序提供超时上下文管理器
项目描述
与asyncio兼容的超时上下文管理器。
使用示例
上下文管理器在需要围绕代码块应用超时逻辑或当 asyncio.wait_for() 不适用时非常有用。此外,它比 asyncio.wait_for() 快得多,因为 timeout 不会创建新的任务。
timeout(delay, *, loop=None) 调用返回一个上下文管理器,当超时到期时会取消该块。
from async_timeout import timeout async with timeout(1.5): await inner()
如果 inner() 在 1.5 秒内执行完成,则不会发生任何操作。
否则,inner() 将由内部通过发送 asyncio.CancelledError 取消,但 asyncio.TimeoutError 会在外部上下文管理器作用域之外抛出。
timeout 参数可以是 None 以跳过超时功能。
或者,可以使用 timeout_at(when) 在绝对时间进行调度。
loop = asyncio.get_event_loop() now = loop.time() async with timeout_at(now + 1.5): await inner()
请注意:这不是 POSIX 时间,而是一个具有未定义起始基数的时间,例如系统启动时间。
上下文管理器有 .expired 属性,用于检查超时是否正好发生在上下文管理器中。
async with timeout(1.5) as cm: await inner() print(cm.expired)
如果 inner() 的执行被超时上下文管理器取消,则该属性为 True。
如果 inner() 调用明确抛出 TimeoutError,则 cm.expired 为 False。
预定截止时间作为 .deadline 属性提供。
async with timeout(1.5) as cm: cm.deadline
可以通过 shift_by() 或 shift_to() 方法重新安排尚未完成的超时。
async with timeout(1.5) as cm: cm.shift(1) # add another second on waiting cm.update(loop.time() + 5) # reschedule to now+5 seconds
如果超时已过期或退出 async with 代码块后,则禁止重新安排。
安装
$ pip install async-timeout
该库仅适用于 Python 3!