retry — Retrying specification object

---

When starting a service, Yellowbox waits for the full initialization of the underlying service (usually docker containers). It does so by polling a function specific for that service. For example, in case of RedisService, Yellowbox attempts to connect to the new Redis server, and start() finishes when it connects successfully.

class retry.RetrySpec(interval: float = 2, attempts: int | None = None, timeout: float | None = None)

Specification object for repeated attempts of an arbitrary action that might fail.

RetrySpec is a dataclass. For arguments, see the defined attributes.

interval

Time in seconds to pause after a failed attempt. Defaults to 2.

attempts

Max number of attempts. If None, infinite attempts are made.

timeout

A timeout for all the attempts (including the interval) combined. If None, function will never time out.

retry(func: Callable[[], T], exceptions: tuple[Type[Exception], ...] | Type[Exception]) → T

Parameters:

• func – A no-argument function that may fail with an exception.

• exceptions – An exception type or tuple of exception types that func can raise and that will trigger a retry.

Retry the given function until it succeeds according to the RetrySpec. Returns the result of the func if it completes successfully. If the maximum number of retries or the timeout is reached, raises the last error raises by func.

Note

If you want to retry a function that takes arguments, use functools.partial() to supply the arguments.

async aretry(func: Callable[[], T], exceptions: tuple[Type[Exception], ...] | Type[Exception]) → T

similar to retry(), but waits asynchronously between attempts.

Example Usage

retry_spec = RetrySpec(interval=0.1, attempts=10)
try:
    # will attempt to get a URL 10 times, with a 0.1 second interval between attempts
    response = retry_spec.retry(lambda: requests.get('https://www.example.com'),
                                RequestException)
except RequestException:
    # all 10 attempts failed
    ...