如何优雅的提高Python应用程序容错性

用户8949263

发布于 2022-04-08 13:26:49

6210

发布于 2022-04-08 13:26:49

前言

如何优雅的提高程序的容错性？

今天分享一种简单可行的方式用来提高 Python 应用程序的稳定性，你是不是立马想到了对代码片段进行重试的改造，我们可以直接使用try ... except ... else语法进行的简单处理，但是更优雅的方式是使用装饰器修饰需要重试的函数。

本文将结合项目实例详细介绍如何实现装饰器多次调用被装饰函数。

一、业务流程函数简介

data_factory函数：被装饰业务数据处理

主要实现数据记录提取解析功能，需要同时提取两条数据记录解析（简单来说就是两个时间点数据计算问题，在允许误差范围内计算出结果），其中一条取最新的数据记录，另外一条为前n小时的记录，这个功能关键问题是前n小时时刻不一定存在数据记录，因此，通过多次执行函数，直到条件满足前返回正常调用结果，确保程序持续稳定运行。

def data_factory(ID,Interval_time,counts):
    """
     #数据处理函数
     Interval_time：时间间隔
     counts：重试次数
     :return:数据处理结果
    """
    # 1、获取数据库数据
    DATA = Data_Extraction(ID)
    DATA['F_DaqDatetime'] = DATA['F_DaqDatetime'].dt.floor('H')
    print(DATA)

    #2、构造测试数据
    DATA=DATA.loc[~DATA["F_DaqDatetime"].isin(["2021-10-23 9:00:00","2021-10-23 10:00:00","2021-10-23 11:00:00"])]
    print("**********2",DATA)
    
    #3、获取时间节点
    #方法一
    in_date1=DATA[["F_DaqDatetime"]].max()[0]
    in_date2=datetime.datetime.strptime(str(in_date1), "%Y-%m-%d %H:%M:%S")
    print("重试次数",counts)
    out_date=in_date2+datetime.timedelta(hours=-(Interval_time+counts))
    print("获取时间点", out_date)
    
    #方法二
    # L_tab=DATA.shape[0]
    # in_date = now_time.strftime('%Y-%m-%d')
    # out_date=pd.date_range(in_date, periods=L_tab, freq='H')[L_tab-(Interval_time+counts)]
    #方法三：数据重采样，适用数据较完整的情况
    
    #4、获取指定时间节点数据
    DATA_0 = DATA[DATA.F_DaqDatetime ==out_date]["F_DaqData"].copy()
    DATA_1 = DATA[DATA.F_DaqDatetime == DATA[["F_DaqDatetime"]].max()[0]]["F_DaqData"].copy()

    print("测试1",DATA_0.values[0])
    print("测试2",DATA_1.values[0])
    
    #业务处理流程此处忽略。。。
    pass
    print("导出完成")

注：

以上数据处理函数主要分为两大部分，数据预处理和后处理，后处理相关的业务流程忽略，主要介绍数据预处理4步骤：

1、获取数据库数据

Data_Extraction(ID)：Data_Extraction（）函数连接数据库提取指定数据源，将F_DaqDatetime非整点时间取整备用(还原重现可塑造类似表结构即可)

2、构造测试数据

我测试时最晚时刻为18时，将以上数据框中9：00、10:00、11:00时数据记录剔除，构造这3个点将取不到数据，覆盖在我取数时间间隔内，将会重试。

3、获取时间节点

分别用3种方法都可以取到需要的数据记录，第一种利用datetime时间模块，第2种通过构造时间序列；第3种采用重采样，前两种时间节点更灵活方便，第3种适用数据完整情况。counts为从装饰器中传递过来的变量值，通过counts变量参数控制移动小时，每重试一次前推1h。

4、获取指定时间节点数据

筛选过滤目标数据记录，提供后续后处理函数解析

5、优化细节思路

进一步提高程序运行效率，每个执行任务函数都可能失败，因此可用加入装饰器拓展函数功能，增加计时、日志记录等，比如一个任务不确定什么时间完成，可设置超时时间，避免占用系统计算资源，如果超时仍然未完成可用通过控制超时重新运行，超过一定次数报错退出。

如果以上容错仍然未成功解析出结果，这时候加入定时重算功能，当数据传输层重传偷偷补上数据记录后，将进一步降低数据计算缺失率。更多装饰器详解见文末推荐阅读。

二、装饰器函数

1）、nonlocal变量，它的作用是函数内部的变量被其修饰后可以使用函数外部对应的局部变量；

2）、try ... except ... else语法，其中else 分支语句的作用是当无异常时进入该分支，有异常的话执行except分支，遇到continue继续下一轮循环，counts变量控制执行次数；对应的还有try ... except ... finally语法，finally的作用是无论最后是否异常都会进入该分支。

3）、如何将装饰器函数中counts参数传递给被装饰函数使用，将重试次数变量存储在关键字字典中，kwargs['运行次数'] = 5 - counts 。

def trying(counts):
    """
     一个装饰器
     传入重试次数
     :return:
     #无固定参数的装饰器（多次重复执行函数，直到返回正常调用结果）
    """
    def Retry_dec(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 使用外部变量
            nonlocal counts
            while counts > 0:
                try:
                    result = func(*args, **kwargs)
                    print('这个函数正常执行:%s' % func.__name__)
                except:
                    counts -= 1
                    print('这个函数出错了:%s' % func.__name__)
                    print('这个函数第%s次运行' %(5-counts))
                    kwargs['运行次数'] = 5 - counts    #将重试次数变量存储,传递给被装饰函数使用

                    continue
                else:
                    return result
        return wrapper

    return Retry_dec

三、测试重试运行效果

将装饰器函数trying()修饰主函数task_process()完成多次调用执行函数，**kwargs传递装饰器中counts参数变量到被装饰函数task_process()中。

**kwargs表示函数接收可变长度的关键字参数字典作为函数的输入。当我们需要函数接收带关键字的参数作为输入的时候，应当使用**kwargs。

@trying(5)
def task_process(**kwargs):
    # 输入待计算ID清单列表
    ID = ["xxxxx"]
    print(kwargs['运行次数'])
    for i in range(len(ID)):
        data_factory(ID[i],Interval_time=5,counts=kwargs['运行次数'])

if __name__ == '__main__':
    task_process()

结果