旁门左道之如何让Python的重试代码更加优雅
By 苏剑林 | 2024-01-14 | 38544位读者 |这篇文章我们讨论一个编程题:如何更优雅地在Python中实现重试。
在文章《新年快乐!记录一下 Cool Papers 的开发体验》中,笔者分享了开发Cool Papers的一些经验,其中就提到了Cool Papers所需要的一些网络通信步骤。但凡涉及到网络通信,就有失败的风险(谁也无法保证网络不会间歇性抽风),所以重试是网络通信的基本操作。此外,当涉及到多进程、数据库、硬件交互等操作时,通常也需要引入重试机制。
在Python中,实现重试并不难,但如何更加简单而又不失可读性地实现重试,还是有一定技巧的。接下来笔者分享一下自己的尝试。
循环重试 #
完整的重试流程大致上包含循环重试、异常处理、延时等待、后续操作等部分,其标准写法就是用for循环,用“try ... except ...”来捕捉异常,一个参考代码是:
import time
from random import random
allright = False # 执行成功的标记
for i in range(5): # 最多重试5次
try:
# 有概率出错的代码
x = random()
if x < 0.5:
yyyy # 未定义yyyy,所以会报错
allright = True
break
except Exception as e:
print(e) # 打印错误信息
if i < 4:
time.sleep(2) # 延时两秒
if allright:
# 执行某些操作
print('执行成功')
else:
# 执行另一些操作
print('执行失败')
接下来我们的目的是简化if allright:
之前的代码,可以发现,它具有比较固定的格式,即在for循环再加上“try ... break ... except ... sleep ...”的模版,不难想象它应该有很大的简化空间。
函数装饰 #
for循环的问题在于,如果有很多处地方需要重试,并且异常处理的方法都一样,那么每次都重写一遍except
的代码就显得很累赘。这种情况下,标准的推荐方法是将有概率出错的代码写成一个函数,并写一个用来处理异常的装饰器:
import time
from random import random
def retry(f):
"""重试装饰器,包装函数加上重试功能
"""
def new_f(*args, **kwargs):
for i in range(5): # 最多重试5次
try:
return True, f(*args, **kwargs)
except Exception as e:
print(e) # 打印错误信息
if i < 4:
time.sleep(2) # 延时两秒
return False, None
return new_f
@retry
def f():
# 有概率出错的代码
x = random()
if x < 0.5:
yyyy # 未定义yyyy,所以会报错
return x
allright, _ = f() # 返回执行状态和执行结果
if allright:
# 执行某些操作
print('执行成功')
else:
# 执行另一些操作
print('执行失败')
当有多处不同的代码都需要重试时,只需要将它们都分别写成函数并加上装饰器@retry
就可以实现相同的重试逻辑,所以装饰器写法确实是一个简明的解决方法,也很直观,所以不难理解能成为标准。目前主流的重试库,比如tenacity,或者更早的retry、retrying等,都是基于装饰器原理的。
理想写法 #
然而,装饰器的写法虽然是标准,但并不完美。首先,需要把重试代码另外封装为一个函数,在很多情况下会让代码显得不够流畅,有种突然卡顿的感觉;其次,由于代码被封装为一个函数,所以代码中的中间变量没法直接用,需要用的变量得全部写在return
中,这显得有点迂回。总的来说,装饰器虽然能够简化重试的代码,但仍然还差点意思。
笔者想象中的完美重试代码,应该是基于上下文管理器的写法,类似于:
with Retry(max_tries=5) as retry:
# 有概率出错的代码
x = random()
if x < 0.5:
yyyy # 未定义yyyy,所以会报错
if retry.allright:
# 执行某些操作
print('执行成功')
else:
# 执行另一些操作
print('执行失败')
然而,笔者去研究了一下上下文管理器的原理之后,才发现这种理想写法注定是无法实现的。因为上下文管理器只能管理上下文,却不能管理主体代码(即本文中那段“有概率出错的代码”)。具体来说,上下文管理器是一个带有__enter__
和__exit__
方法的类,它将__enter__
插入到代码运行之前(上文),将__exit__
插入到代码运行之后(下文),但无法操控中间的代码(比如让它运行多几次)。
所以,基于上下文管理器的一行实现重试的写法宣告失败。
挣扎一下 #
不过,好消息是,上下文管理器虽然不能实现循环,但它的__exit__
方法能处理异常,所以它至少能代替“try ... except ...”来处理异常,于是我们可以写出:
import time
from random import random
class Retry:
"""自定义处理异常的上下文管理器
"""
def __enter__(self):
self.allright = False
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is None:
self.allright = True
else:
print(exc_val)
time.sleep(2)
return True
for i in range(5): # 最多重试5次
with Retry() as retry:
# 有概率出错的代码
x = random()
if x < 0.5:
yyyy # 未定义yyyy,所以会报错
break
if retry.allright:
# 执行某些操作
print('执行成功')
else:
# 执行另一些操作
print('执行失败')
最新的这个版本,其实写法上已经非常接近前一节的理想写法了,不同点有两个:1、需要多写一句for
循环,但这个无法避免,因为前面已经说了上下文管理器是无法启动循环的,所以只能自己额外用for
或者while
启动循环;2、需要自己显式加一句break
,这个倒是可以想办法优化掉。
此外,这个版本还有一个小缺陷,就是假如所有重试都失败了,那么最后一次重试失败之后,依然会启动sleep,这理论上是没有必要的,应当想办法去掉。
继续优化 #
为了优化掉break
,那么模型的循环应该要学会自己停止,这有两个办法:第一个办法是可以改用while
循环,然后让停止条件根据重试结果改变,这将会导致跟《Handling exceptions inside context managers》相近的结果;第二个办法则是保持for
循环,但range(5)
这个要换成根据重试结果变化的迭代器。本文主要探讨后一种方案。
经过分析,笔者发现可以通过内置方法__call__
和__iter__
将retry
同时作为一个可变的迭代器,并解决最后一次失败后的非必要sleep问题:
import time
from random import random
class Retry:
"""处理异常的上下文管理器 + 迭代器
"""
def __call__(self, max_tries=5):
self.max_tries = max_tries
return self
def __iter__(self):
for i in range(self.max_tries):
yield i
if self.allright or i == self.max_tries - 1:
return
time.sleep(2)
def __enter__(self):
self.allright = False
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is None:
self.allright = True
else:
print(exc_val)
return True
retry = Retry()
for i in retry(5): # 最多重试5次
with retry:
# 有概率出错的代码
x = random()
if x < 0.5:
yyyy # 未定义yyyy,所以会报错
if retry.allright:
# 执行某些操作
print('执行成功')
else:
# 执行另一些操作
print('执行失败')
仔细对比的读者可能会疑问:你想办法删了一句break
,但多了一句retry = Retry()
,总的代码数不变(还把上下文管理器搞复杂了),有这个必要折腾吗?事实上这里的retry
是可重用的,用户只需要一次定义retry = Retry()
,那么在后面的重试中都只需要:
for i in retry(max_tries):
with retry:
# 有概率出错的代码
就行了,因此虽然把上下文的管理器搞复杂了一点,但却已经是无限接近理想写法的实现了。
终极版本 #
不过,“一次定义retry = Retry()
,多次重用retry
”只适合于单进程,如果是多进程还是要分别定义retry = Retry()
的,另外就是这样重用总给人“不同的重试之间似乎没有完全隔离”的感觉。有没有可能将这一句完全去掉呢?笔者再想一下,发现还是有可能的!参考代码如下:
import time
from random import random
class Retry:
"""处理异常的上下文管理器 + 迭代器
"""
def __init__(self, max_tries=5):
self.max_tries = max_tries
def __iter__(self):
for i in range(self.max_tries):
yield self
if self.allright or i == self.max_tries - 1:
return
time.sleep(2)
def __enter__(self):
self.allright = False
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is None:
self.allright = True
else:
print(exc_val)
return True
for retry in Retry(5): # 最多重试5次
with retry:
# 有概率出错的代码
x = random()
if x < 0.5:
yyyy # 未定义yyyy,所以会报错
if retry.allright:
# 执行某些操作
print('执行成功')
else:
# 执行另一些操作
print('执行失败')
这次的改动是将__call__
换为__init__
,然后__iter__
中的yield i
改为了yield self
,即返回对象本身。这样一来,就不用单独写一行retry = Retry(5)
来初始化,而是在for retry in Retry(5):
中同时实现了初始化和别名赋值,并且由于每次重试都会重新初始化,因此实现了重试之间的完全隔离,可谓一举两得了。
文章小结 #
本文相对完整地探讨了Python中重试机制的写法,试图得到笔者心目中重试代码的完美实现,最后的结果也勉强达到了心中所想。
不过,不得不承认的是,这篇文章的起因实质只是“强迫症”在作祟,并没有算法效率的实质改进,在编程细节上花太多时间,某种程度上已经是“旁门左道”、“不务正业”了,并不是一件十分值得学习的事情。
转载到请包括本文地址:https://www.kexue.fm/archives/9938
更详细的转载事宜请参考:《科学空间FAQ》
如果您还有什么疑惑或建议,欢迎在下方评论区继续讨论。
如果您觉得本文还不错,欢迎分享/打赏本文。打赏并非要从中获得收益,而是希望知道科学空间获得了多少读者的真心关注。当然,如果你无视它,也不会影响你的阅读。再次表示欢迎和感谢!
如果您需要引用本文,请参考:
苏剑林. (Jan. 14, 2024). 《旁门左道之如何让Python的重试代码更加优雅 》[Blog post]. Retrieved from https://www.kexue.fm/archives/9938
@online{kexuefm-9938,
title={旁门左道之如何让Python的重试代码更加优雅},
author={苏剑林},
year={2024},
month={Jan},
url={\url{https://www.kexue.fm/archives/9938}},
}
January 14th, 2024
[...]Read More [...]
January 15th, 2024
确实有一个这样的库,jd/tenacity,同样也是有装饰器和上下文管理器两个选择。
搜了搜,原来 tenacity 确实已经支持了相同的写法:https://tenacity.readthedocs.io/en/latest/#retrying-code-block 。但我相当不理解的是,为什么这种写法没有成为主流,而是装饰器成为主流。
装饰器不用改代码呀,直接在已实现的函数上加一个装饰器就行
问题在于,通常来说并不是整个函数的所有代码都需要重试,很多时候只有几行甚至一两行代码需要重试。
January 23rd, 2024
from tenacity import wait_random, AsyncRetrying, stop_after_attempt
同@苏剑林|comment-23504,后面我也发现tenacity支持同样的
for retry in Retry(5): # 最多重试5次
with retry:
写法了,不过还是自己发现一遍更清晰,也更方便定制化。
January 29th, 2024
可以实现一个类似于torch.no_grad(),即可以作为decorator也可以作为context manager来使用?
以前写过一个这样的东西,不过是用来对代码片段做计时的:https://github.com/typoverflow/UtilsRL/blob/6e30fc4f0b524516089de4df1662e31ec498093a/UtilsRL/misc/decorator.py#L63-L148
可以的,增加相应的__call__方法,就可以将类当装饰器用了
February 3rd, 2024
语法上变不了就语义上变一下呗,例如这样:
for ctx in CTX():
with ctx.retry(5):
x = random()
if x < 0.5:
yyyy # 未定义yyyy,所以会报错
@ctx.set_except
def func(e):
pass
@ctx.set_final
def func():
nonlocal x
del x
# execpt/final在__iter__时才触发