背景

在Python自动化中,我们经常需要对某个数据库表进行轮询查询来验证某张表在我们做了某个操作后数据库是否新增了一条数据或者某条数据发生了更新,此时我们需要用数据库轮询的方式来对结果进行查询

具体实现代码

轮询主体

def poll_function(func, stop_condition,interval=1, max_attempts=5,is_assert='N', *args, **kwargs):
    """
    通用数据库轮询方法：执行一个查数据库函数，直到满足停止条件或达到最大尝试次数。（使用方法参考下方main方法）
    :param func: 要执行的函数。
    :param stop_condition: 停止函数，返回True时停止轮询。
        当只需要判断查询的值是否存在时参数录入lambda result: my_stop_condition(result)
        当需要判断查询的值里某个字段是否对应某个值时录入lambda result: my_stop_condition(result,expectData={'预期字段':'预期值'})，比如要判断ANS_COD字段是否为SU00时my_stop_condition(result,expectData={'ANS_COD':'SU00'})
    :param interval: 轮询间隔时间（秒）。
    :param max_attempts: 最大尝试次数。
    :param is_assert: 是否断言
    :param args: 函数的位置参数。
    :param kwargs: 函数的关键字参数。
    """
    attempts = 0
    while attempts < max_attempts:
        result = func(*args, **kwargs)
        if stop_condition(result):
            logger.info("【数据库通用轮询】满足停止条件，结束轮询。最终结果:{}".format(result))
            return result
        logger.info("【数据库通用轮询】尝试 #{}: {}".format(attempts + 1,result))
        time.sleep(interval)
        attempts += 1
    if is_assert == 'Y':
        # is_assert为Y时，当轮询结束都不符合停止条件直接断言结束代码运行。
        assert False, f"【数据库通用轮询】达到最大尝试次数 {max_attempts}，未满足停止条件。"
    else:
        # is_assert为N时，当轮询结束都不符合停止条件返回[]，不直接停止代码运行，可以尝试去别的表查询。
        return []

停止条件

def my_stop_condition(dbResult=[],expectData={}):
    '''
    停止条件
    :param dbResult: 数据库查询结果，固定把result传给它就行
    :param expectData: 如果需要判断查询结果里某个字段是否对应某个值时，需要填，格式{'字段值':'预期值'}，只需要轮询返回的结果不为[]就行时，这个字段不填
    :return:
    '''
    if expectData in ('',{},None):
        # 只判断轮询结果是否非空
        if dbResult != []:
            return True
        else:
            return False
    else:
        # 轮询查询结果里某个字段是否对应指定的某个预期值
        return CommonTool().commonDataCheck(dbResult, expectData);