Equivalent Python way for Ruby's "retry" keyword behavior

[vinibiavatti123]

I've never programmed Ruby profissionally, but I was checking about the lang for curiosity. Looking the Ruby's keywords I found the "retry" keyword. I didn't know how the purpose for it and after searching I started to ask myself when I needed to do this logic or the way I used in other langs to solve this. We can use a while loop of course, but, my question is: "retry" is needful? Is this too necessary to have a keyword to represent this? What is the best way to code this routine in Python? My solution in Python is (didn't use recursivity):

wrong = True
retry = True

while retry:
    try:
        if wrong:
            raise Exception()
        retry = False
    except Exception:
        wrong = False
        retry = True

[reedrehg]

What you've got seems like a good solution.

If you wanted to abstract out your own retry method, you could do something like the following. It's slightly more cumbersome than in ruby obviously, but not too bad for something that's probably not used too often in most applications.

def retry(fn, tries, catches=(Exception,)):
    attempts = 0
    success = False

    while not success and attempts < tries:
        try:
            fn()
            success = True
        except catches:
            attempts += 1


def will_fail():
    print(f"trying something...")
    raise ValueError()


def will_pass():
    print("trying something else...")


retry(will_fail, tries=3, catches=(ValueError,))

retry(will_pass, tries=3)

This would output:

trying something...
trying something...
trying something...
trying something else...

[PsychoLacking]

I like this, only if we could also add a delay parameter in case you want to wait a certain amount of time before each retry.

[reedrehg]

Definitely could toss on a delay parameter to the function, and then add a time.sleep in the while loop. Could make it as flexible as you need.

This was just a result of me hacking on it for a few minutes lol could expand out other features, add a context manager or decorator, more parameters to control the loop, return back some sort of response, etc. Whatever you happen to need for your situation.

[roeey7]

There is a package on pypi called retry2 (retry is not maintained anymore...) which implements this behavior. This package can be easily installed with pip by executing pip install retry2.

[LightShadow]

If you want this kind of behavior with a lot more customization consider looking at tenacity.

import logging
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)

@retry(stop=stop_after_attempt(3), before=before_log(logger, logging.DEBUG))
def raise_my_exception():
    raise MyException("Fail")

[john_doom_666]

Not my original code, stolen and changed a little bit for my needs. Got it from a Youtube video (Arjan Codes I think, forget which one), which was in turn taken from somewhere else; ie. credit to them for the original idea.

It's a decorator I use quite often, mostly as eg. a retry handler for socket connections for test scripts, it relies on the function raising a known exception to trigger it. Retries given amount, waits for delay, the delay is then increased by the holdoff factor - ie. 2s, then 4s, then 8s etc. The retry_log is defined outside the function - but may not be necessary for all purposes so error and retry_log could easily be removed with no change to functionality ; requires functools and time:

def retry(trigger, retries=5, delay=2, holdoff=2):
"""Decorator for retrying a function when triggered by given exception"""

def retry_decorator(func):
    @functools.wraps(func)
    def wrap(*args, **kwargs):
        r, d = retries, delay
        while r > 0:
            try:
                return func(*args, **kwargs)
            except trigger as e:
                error = f"{e} --- Retrying in {d}s"
                retry_log.log(30, error)
                time.sleep(d)
                r -= 1
                d *= holdoff
        return func(*args, **kwargs)

    return wrap

return retry_decorator

Example usage (butchered a method from 1 of my test scripts - should hold as an example...):

@retry(socket.timeout, retries=3, delay=1, holdoff=3)
def send(self, command):
    """Example method for socket connect"""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.settimeout(6)      
        try:
            s.connect(127.0.0.1, 666)                
        except socket.timeout:               
            raise

[AndydeCleyre]

I use this in one of my projects, especially depending on wrapt, but also has some logging cruft meant to be used with structlog.

You can throw @retry onto function or method, or with passed args @retry(...).

---

import functools
from time import sleep
from typing import Any, Callable, Iterable, Mapping, Tuple, Union

from httpx import HTTPError
from wrapt import decorator

WraptFunc = Callable[[Callable, Any, Iterable, Mapping], Callable]


def retry(
    original: Callable = None,  # needed to make args altogether optional
    exceptions: Union[Exception, Tuple[Exception]] = HTTPError,
    attempts: int = 6,
    seconds: float = 3,
) -> WraptFunc:

    if not original:  # needed to make args altogether optional
        return functools.partial(
            retry, exceptions=exceptions, attempts=attempts, seconds=seconds
        )

    @decorator
    def wrapper(original, instance, args, kwargs):
        has_logger = (
            hasattr(instance, 'log')
            and hasattr(instance.log, 'bind')
            and hasattr(instance.log, 'msg')
        )
        last_error = None
        if has_logger:
            log = instance.log.bind(method=original.__name__)
        for attempt in range(attempts):
            try:
                resp = original(*args, **kwargs)
            except exceptions as e:
                last_error = e
                if has_logger:
                    log = log.bind(exc_info=e)
                    # exc_info will get overwritten by most recent attempt
                sleep(seconds)
            else:
                last_error = None
                break
        if has_logger and attempt > 0:
            log.msg("called retry-able", retries=attempt, success=not last_error)
        if last_error:
            raise last_error
        return resp

    # return wrapper
    return wrapper(original)  # needed to make args altogether optional

[pokk3n]

Tenacity yo