tenacity发生异常/失败/错误时重试retry机制,Python

x33g5p2x  于2022-08-17 转载在 Python  
字(2.4k)|赞(0)|评价(0)|浏览(812)

tenacity发生异常/失败/错误时重试retry机制,Python

安装:

  1. pip install tenacity

示例:

  1. @retry
  2. def non_stop():
  3. print("永不停息")
  4. raise Exception

代码运行后,将用不停息的输出:

  1. 永不停息
  2. 永不停息
  3. 永不停息
  4. ...

发生异常重试若干次后退出。

  1. from tenacity import retry, stop_after_attempt
  2. @retry(stop=stop_after_attempt(3))
  3. def non_stop():
  4. print("永不停息")
  5. raise Exception

输出:

  1. 永不停息
  2. 永不停息
  3. 永不停息

上面代码发生异常重试3次后结束重试。

重试若干时间(秒)后退出:

  1. import datetime
  2. from tenacity import retry, stop_after_delay
  3. @retry(stop=stop_after_delay(2))
  4. def non_stop():
  5. t = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
  6. print(t)
  7. raise Exception

上面代码发生异常错误重试2秒后结束重试。

发生异常重试若干次或者重试若干秒后结束:

  1. import datetime
  2. from tenacity import retry, stop_after_delay, stop_after_attempt
  3. @retry(stop=(stop_after_delay(2) | stop_after_attempt(3)))
  4. def retry_stop():
  5. t = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
  6. print(t)
  7. raise Exception

上面代码重试3次或者重试2秒后结束。

发生异常后,等待一定时间后重试:

  1. import datetime
  2. from tenacity import retry, wait_fixed
  3. @retry(wait=wait_fixed(3))
  4. def retry_stop():
  5. t = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
  6. print(t)
  7. raise Exception

以上代码发生异常后,等待3秒重试。

发生异常,等待一定随机时间后重试:

  1. import datetime
  2. from tenacity import retry, wait_random
  3. @retry(wait=wait_random(min=2, max=5))
  4. def exc_retry():
  5. t = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
  6. print(t)
  7. raise Exception

以上代码,发生异常后,等待最少2秒,最多5秒的随机时间后重试。

发生异常后,等待一定量时间,再加上一定随机时间后重试:

  1. import datetime
  2. from tenacity import retry, wait_random, wait_fixed
  3. @retry(wait=wait_fixed(2) + wait_random(1, 3))
  4. def exc_retry():
  5. t = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
  6. print(t)
  7. raise Exception

上面代码发生异常后,等待2秒的基础上再加上1到3秒之间随机时间总和后重试。

只有当发生特定异常后才重试:

  1. import datetime
  2. from tenacity import retry, retry_if_exception_type
  3. @retry(retry=retry_if_exception_type(IOError))
  4. def exc_retry():
  5. t = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
  6. print(t)
  7. raise IOError

上面代码,只有发生IOError后才重试。

只有当满足条件为True时候,才重试:

  1. import datetime
  2. from tenacity import retry, retry_if_result
  3. def is_value(value):
  4. print(value)
  5. return True
  6. @retry(retry=retry_if_result(is_value))
  7. def program_retry():
  8. t = datetime.datetime.now().strftime('%M:%S')
  9. print(t)
  10. return -1
  11. # raise Exception

program_retry()返回的值进入is_value()函数处理,只有当is_value()函数返回值为True时候才重试,如果is_value()返回为False,则停止重试。

查看异常信息统计:

  1. import datetime
  2. from tenacity import retry, stop_after_attempt
  3. @retry(stop=stop_after_attempt(3))
  4. def program_retry():
  5. t = datetime.datetime.now().strftime('%M:%S')
  6. print(t)
  7. raise Exception()
  8. if __name__ == '__main__':
  9. try:
  10. program_retry()
  11. except:
  12. pass
  13. print(program_retry.retry.statistics)

输出:

  1. 23:32
  2. 23:32
  3. 23:32
  4. {'start_time': 你的程序启动时间, 'attempt_number': 3, 'idle_for': 0, 'delay_since_first_attempt': 0.0}

相关文章