A retry backoff strategy is an algorithm that varies the backoff interval
between retries. If you do not want to use one of the builtin common backoff
retry strategies, you
can implement your own by deriving a callable from
BackoffBase, which lives in
For real life inspiration, you can see all of pypyr’s builtin retry backoff implementations on github.
You can find
class BackoffBase(abc.ABC): """Derive from me for a custom back-off strategy. Attributes: sleep (float): initial sleep interval in seconds. max_sleep (float): upper bound for sleep interval calculation. jrc (float): Jitter Range Coefficient. kwargs (dict): arbitrary keyword args for use in deriving classes. """ def __init__(self, sleep=1, max_sleep=None, jrc=0, kwargs=None): """Initialize back-off strategy. Args: sleep (float): positive sleep interval in seconds. max_sleep (float): upper bound for sleep interval calculation. jrc (float): jitter randomizes between [sleep*jrc] and [sleep]. kwargs (dict): arbitrary arguments for use by deriving classes. """ self.sleep = sleep self.max_sleep = max_sleep self.jrc = jrc self.kwargs = kwargs @abc.abstractmethod def __call__(self, n): """Implement back-off sleep interval calculation here. Args: n (int): The iteration counter. Returns: Float. The sleep interval for iteration n. """ raise NotImplementedError() def min(self, sleep): """Return the smaller of sleep vs max_sleep.""" max_sleep = self.max_sleep return min(sleep, max_sleep) if max_sleep else sleep def randomize(self, sleep): """Get random number in between (jrc*sleep) and (sleep). Does NOT guard against result < max_sleep. This is on purpose. It allows you to jitter under or over the sleep interval depending on if jrc > 1. """ return random.uniform(sleep*self.jrc, sleep)
import pypyr.retries class MyBackoff(pypyr.retries.BackoffBase): """Your own custom backoff algorithm.""" def __call__(self, n): """Just return n + sleep + custom input myArg.""" # probably do something more useful here. # myArg comes from backoffArgs you set in pipeline. # this will sleep for: # (iteration count + sleep + myArg) in seconds. return n + self.sleep + self.kwargs['myArg']
You can use
self.min(my_interval) to honor
self.randomize(my_interval) to add jitter with
jrc in the same way the
builtin retry algorithms do.
You can use your custom backoff in a pipeline by setting the
on the step’s retry decorator.
Specify your custom backoff callable by using its containing module’s absolute name followed by the class name.
To load a custom callable, the name should be in format
The usual python import module resolution rules apply. pypyr will resolve
modules from the
first, which you can change by using the
Assuming your callable class
MyBackoff exists in a file like this
./dir/mybackoffs.py, you can use use it in your pipeline like this:
- name: pypyr.steps.cmd retry: backoff: dir.mybackoffs.MyBackoff max: 123 sleep: 456 backoffArgs: myArg: 789.01 in: cmd: curl xyz://manifestly-wrong-url
If you package your code and you install the package into the active python environment, you can of course use the usual python package name instead, followed by the class name:
- name: pypyr.steps.cmd retry: backoff: mypackage.mymodule.MyBackoff max: 123 sleep: 456 backoffArgs: myArg: 789.01 in: cmd: curl xyz://manifestly-wrong-url