使用Python的tenacity库实现异常重试机制

当前位置:

首页 > temp > python入门教程 >

使用Python的tenacity库实现异常重试机制

Python异常重试
如果某些方法需要重试，之前都是自己在except里写代码，很多时候实现起来并不方便，而且复杂逻辑下大大增加代码量和影响可读性，尝试了一下tenacity库，重试机制非常简单易用且清晰，且比retrying库简洁很多。
1，第一种常规方式，不借助第三方库，需要自行再except里写代码，真个方法显得十分臃肿
2，第二种方法使用tenacity库，实现起来非常优雅、轻便，当然tenacity库有非常强大的功能，有必要再探索。
demo一下数据库连接的异常重试实现，终于不用再自己写臃肿的重试逻辑了。

import time

import psycopg2
import datetime
from tenacity import *

g_attempt = 0


# 直接再异常中进行代码的重试，需要一个global的变量做计数器，
# 当方法执行发生异常后重试5次，每次间隔10秒
def my_retry_function_v1(conn):
    global g_attempt
    conn = None
    try:
        conn = psycopg2.connect(**conn)
        print('my_retry_function execute successfully')
        g_attempt = 0
    except Exception as err:
        g_attempt = g_attempt + 1
        if g_attempt <= 5:
            print(str(datetime.datetime.now()) + f' :retry---> {str(err)} ')
            print(f'retrying: {str(g_attempt)}')
            time.sleep(10)
            return my_retry_function_v1(conn)
        else:
            #记录异常，这里仅用print示例
            print('sent a alert notification')

    if conn:
        return conn


#方法执行发生异常后重试5次，每次间隔10秒
@retry(stop=stop_after_attempt(5), wait=wait_fixed(10))
def my_retry_function_v2(conn):
    conn = None
    try:
        conn = psycopg2.connect(**conn)
        print('my_retry_function execute successfully')
    except Exception as err:
        print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f') + f' :retry---> {str(err)} ')
        #达到最大重试之后，记录异常信息
        if (my_retry_function_v2.retry.statistics['attempt_number']) == 5:
            #记录异常，这里仅用print示例
            print('sent a alert notification')
        else:
            #未超出最大重试次数之前，显式抛出异常,继续重试
            raise
    if conn:
        return conn


if __name__== '__main__':
    conn = {'host': '*', 'database': 'postgres', 'user': 'postgres', 'password': '*','port':*}
    my_retry_function_v1(conn)

出处：https://www.cnblogs.com/wy123/p/15093729.html

栏目列表