这篇文章我们讨论一个编程题:如何更优雅地在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,或者更早的retryretrying等,都是基于装饰器原理的。

理想写法 #

然而,装饰器的写法虽然是标准,但并不完美。首先,需要把重试代码另外封装为一个函数,在很多情况下会让代码显得不够流畅,有种突然卡顿的感觉;其次,由于代码被封装为一个函数,所以代码中的中间变量没法直接用,需要用的变量得全部写在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}},
}