Source code for riprova.strategies.constant

# -*- coding: utf-8 -*-
from ..backoff import Backoff
from ..constants import INT_ERROR, POS_ERROR

[docs]class ConstantBackoff(Backoff): """ ConstantBackOff is a backoff policy that always returns the same backoff delay. The constant backoff policy can be configured with a custom retry interval and maximum number of retries. This is in contrast to an exponential backoff policy, which returns a delay that grows longer as you call `next()`. ConstantBackoff is expected to run in a single-thread context. Arguments: interval (int|float): wait interval before retry in seconds. Use `0` for no wait. Defaults to `0.1` = `100` milliseconds. retries (int): maximum number of retries attempts. Use `0` for no limit. Defaults to `10`. Raises: AssertionError: in case of invalid params. Usge:: @riprova.retry(backoff=riprova.ConstantBackoff(retries=5)) def task(x): return x * x """ def __init__(self, interval=.1, retries=10): assert isinstance(retries, int), INT_ERROR.format('retries') assert isinstance(interval, (int, float)), INT_ERROR.format('interval') assert retries >= 0, POS_ERROR.format('retries') assert interval >= 0, POS_ERROR.format('interval') self.retries = retries self.pending_retries = retries self.interval = interval
[docs] def reset(self): """ Resets the current backoff state data. """ self.pending_retries = self.retries
[docs] def next(self): """ Returns the number of seconds to wait before the next try, otherwise returns `Backoff.STOP`, which indicates the max number of retry operations were reached. Returns: float: time to wait in seconds before the next try. """ # Verify we do not exceeded the max retries if self.retries > 0 and self.pending_retries == 0: return Backoff.STOP # Decrement pending retries attempts if self.retries > 0: self.pending_retries -= 1 # Return pending interval return self.interval