使用Python装饰器重试请求

xzlaal3s  于 2023-06-28  发布在  Python
关注(0)|答案(8)|浏览(184)

我在我的脚本中有多个函数,它执行REST API API请求。由于我需要处理错误情况,我已经将重试机制如下所示。

  1. no_of_retries = 3
  2. def check_status():
  3. for i in range(0,no_of_retries):
  4. url = "http://something/something"
  5. try:
  6. result = requests.get(url, auth=HTTPBasicAuth(COMMON_USERNAME, COMMON_PASSWORD)).json()
  7. if 'error' not in result:
  8. return result
  9. else:
  10. continue
  11. except Exception as e:
  12. continue
  13. return None

我有几种不同的方法做类似的操作。如何才能更好的避免重复的方法可能是使用装饰器。

iq0todco

iq0todco1#

如果你不介意安装一个库,你可以使用tenacitygithub.com/jd/tenacity)模块。他们的一个例子:

  1. import random
  2. from tenacity import retry, stop_after_attempt
  3. # @retry # retry forever
  4. @retry(stop=stop_after_attempt(3))
  5. def do_something_unreliable():
  6. if random.randint(0, 10) > 1:
  7. raise IOError("Broken sauce, everything is hosed!!!111one")
  8. else:
  9. return "Awesome sauce!"
  10. print(do_something_unreliable())

这也允许你指定你想要保持重试的尝试次数或秒数。
对于您的情况,这可能看起来像这样(未测试!):

  1. @retry(stop=stop_after_attempt(3))
  2. def retry_get():
  3. result = requests.get(
  4. url, auth=HTTPBasicAuth(COMMON_USERNAME, COMMON_PASSWORD)).json()
  5. if 'error' not in result:
  6. raise RequestException(result)
展开查看全部
ibrsph3r

ibrsph3r2#

您可以使用这样的装饰器并处理自己的异常。

  1. def retry(times, exceptions):
  2. """
  3. Retry Decorator
  4. Retries the wrapped function/method `times` times if the exceptions listed
  5. in ``exceptions`` are thrown
  6. :param times: The number of times to repeat the wrapped function/method
  7. :type times: Int
  8. :param Exceptions: Lists of exceptions that trigger a retry attempt
  9. :type Exceptions: Tuple of Exceptions
  10. """
  11. def decorator(func):
  12. def newfn(*args, **kwargs):
  13. attempt = 0
  14. while attempt < times:
  15. try:
  16. return func(*args, **kwargs)
  17. except exceptions:
  18. print(
  19. 'Exception thrown when attempting to run %s, attempt '
  20. '%d of %d' % (func, attempt, times)
  21. )
  22. attempt += 1
  23. return func(*args, **kwargs)
  24. return newfn
  25. return decorator
  26. @retry(times=3, exceptions=(ValueError, TypeError))
  27. def foo1():
  28. print('Some code here ....')
  29. print('Oh no, we have exception')
  30. raise ValueError('Some error')
  31. foo1()
展开查看全部
ttygqcqt

ttygqcqt3#

第三方retry module是用于此目的的选项。您还可以传递要重试的异常列表、重试次数、延迟、最大延迟、指数回退等。

  1. $ pip install retry

示例用法:

  1. from retry import retry
  2. @retry(ZeroDivisionError, tries=3, delay=2)
  3. def make_trouble():
  4. '''Retry on ZeroDivisionError, raise error after 3 attempts, sleep 2 seconds between attempts.'''
im9ewurl

im9ewurl4#

生产级示例

  1. import logging
  2. import time
  3. import functools
  4. import traceback
  5. LOG_FORMAT = "%(asctime)s - %(levelname)s - %(pathname)s - %(funcName)s - %(lineno)d -msg: %(message)s"
  6. logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
  7. def retry(retry_num, retry_sleep_sec):
  8. """
  9. retry help decorator.
  10. :param retry_num: the retry num; retry sleep sec
  11. :return: decorator
  12. """
  13. def decorator(func):
  14. """decorator"""
  15. # preserve information about the original function, or the func name will be "wrapper" not "func"
  16. @functools.wraps(func)
  17. def wrapper(*args, **kwargs):
  18. """wrapper"""
  19. for attempt in range(retry_num):
  20. try:
  21. return func(*args, **kwargs) # should return the raw function's return value
  22. except Exception as err: # pylint: disable=broad-except
  23. logging.error(err)
  24. logging.error(traceback.format_exc())
  25. time.sleep(retry_sleep_sec)
  26. logging.error("Trying attempt %s of %s.", attempt + 1, retry_num)
  27. logging.error("func %s retry failed", func)
  28. raise Exception('Exceed max retry num: {} failed'.format(retry_num))
  29. return wrapper
  30. return decorator

用法

  1. # this means try your function 5 times, each time sleep 60 seconds
  2. @retry(5, 60)
  3. def your_func():
  4. pass

正式参考:https://peps.python.org/pep-0318/

展开查看全部
klr1opcd

klr1opcd5#

我推荐使用retry库,就像@MohitC提到的那样。但是,如果你像我一样被限制导入第三方库,欢迎你尝试我的版本:

  1. import time
  2. def retry(tries= -1, delay=0, max_delay=None, backoff=1, exceptions=Exception, log=False):
  3. """Retry Decorator with arguments
  4. Args:
  5. tries (int): The maximum number of attempts. Defaults to -1 (infinite)
  6. delay (int, optional): Delay between attempts (seconds). Defaults to 0
  7. max_delay (int, optional): The maximum value of delay (seconds). Defaults to None (Unlimited)
  8. backoff (int, optional): Multiplier applied to delay between attempts (seconds). Defaults to 1 (No backoff)
  9. exceptions (tuple, optional): Types of exceptions to catch. Defaults to Exception (all)
  10. log (bool, optional): Print debug logs. Defaults to False
  11. """
  12. def retry_decorator(func):
  13. def retry_wrapper(*args, **kwargs):
  14. nonlocal tries, delay, max_delay, backoff, exceptions, log
  15. while tries:
  16. try:
  17. return func(*args, **kwargs)
  18. except exceptions:
  19. tries -= 1
  20. # Reached to maximum tries
  21. if not tries:
  22. raise
  23. # Log the retry logs for the given function
  24. if log:
  25. print(f"Retrying {func.__name__} in {delay} seconds")
  26. # Apply delay between requests
  27. time.sleep(delay)
  28. # Adjust the next delay according to backoff
  29. delay *= backoff
  30. # Adjust maximum delay duration
  31. if max_delay is not None:
  32. delay = min(delay, max_delay)
  33. return retry_wrapper
  34. return retry_decorator

使用示例:

很简单:

  1. @retry(10, delay=5)
  2. def do_something(params):
  3. # Example func to retry
  4. pass

高级:

  1. @retry(10, delay=1, backoff=2, max_delay=10, exceptions=(TimeoutError), log=True)
  2. def do_something(params):
  3. # Example func to retry only for TimeoutErrors
  4. pass
展开查看全部
mbjcgjjk

mbjcgjjk6#

与其使用装饰器,可能更好的解决方案是将请求移动到它自己的函数中,得到类似于下面的结构:

  1. no_of_retries = 3
  2. def make_request(url):
  3. for i in range(0,no_of_retries):
  4. try:
  5. result = requests.get(url, auth=HTTPBasicAuth(COMMON_USERNAME, COMMON_PASSWORD)).json()
  6. if 'error' not in result:
  7. return result
  8. else:
  9. continue
  10. except Exception as e:
  11. continue
  12. return result
  13. def check_status():
  14. result = make_request("http://something/status")
  15. def load_file():
  16. result = make_request("http://something/file")

这样,在封装请求时可以避免重复代码。如果你要使用装饰器,你需要 Package 整个load_file()方法,这将阻止你在这个函数中进一步处理请求的结果。

展开查看全部
wr98u20j

wr98u20j7#

我创建了一个自定义重试函数。如果第一个数字小于第二个数字,则该函数将重试。

验证码:

  1. import time
  2. # CUSTOM EXCEPTION
  3. class custom_error(Exception):
  4. pass
  5. # RETRY FUNCTION.
  6. def retry(func, retries=3):
  7. print(func)
  8. def retry_wrapper(*args, **kwargs):
  9. print(args)
  10. n = args[0]
  11. u = args[1]
  12. print(n, u)
  13. attempts = 0
  14. while attempts < retries:
  15. try:
  16. if n > u:
  17. return func(*args, **kwargs)
  18. else:
  19. raise custom_error
  20. except custom_error:
  21. print("error")
  22. time.sleep(2)
  23. attempts += 1
  24. return retry_wrapper
  25. @retry
  26. def akash(a, b):
  27. c = a / b
  28. return c
  29. # CALLING THE FUNCTION
  30. a = akash(1, 2)
  31. print(a)

输出:

  1. <function akash at 0x00000187C3A66B00>
  2. (1, 2)
  3. 1 2
  4. error
  5. error
  6. error
展开查看全部
vojdkbi0

vojdkbi08#

在mrkiril的答案上使用functools

  1. from functools import wraps, partial
  2. def retry(f=None, times=10):
  3. if f is None:
  4. return partial(retry, times=times)
  5. @wraps(f)
  6. def wrap(*args, **kwargs):
  7. attempt = 0
  8. while attempt < times:
  9. try:
  10. return f(*args, **kwargs)
  11. except:
  12. print(f"{f.__name__}, attempt {attempt} of {times}")
  13. attempt += 1
  14. return f(*args, **kwargs)
  15. return wrap

然后,像下面这样 Package 你的函数:

  1. import random
  2. @retry
  3. def foo():
  4. if random.randint(0, 5) != 0:
  5. raise Exception
展开查看全部

相关问题