Skip to content

Instantly share code, notes, and snippets.

@rqx110
Created April 3, 2020 07:16
Show Gist options
  • Save rqx110/5dc910a858869a8b70256da238516b91 to your computer and use it in GitHub Desktop.
Save rqx110/5dc910a858869a8b70256da238516b91 to your computer and use it in GitHub Desktop.
retry operation
class Retrying:
def __init__(self, interval=1.0, delay=0.05, maxAttempts=None):
self.interval = interval
self.delay = delay
self.maxAttempts = maxAttempts
def __iter__(self):
class Iterator:
def __init__(self, interval, delay, maxAttempts):
self.attempt = 0
self.startedAt_ = datetime.now()
self.interval_ = interval
self.delay_ = delay
self.maxAttempts_ = maxAttempts
def __next__(self):
time.sleep(self.delay_)
if self.isExpired() or \
self.maxAttempts_ and self.attempt >= self.maxAttempts_:
raise StopIteration
self.attempt += 1
return self
def next(self):
return self.__next__()
def isExpired(self):
return self.interval_ and \
(datetime.now() - self.startedAt_).total_seconds() > self.interval_
return Iterator(self.interval, self.delay, self.maxAttempts)
@rqx110
Copy link
Author

rqx110 commented Apr 3, 2020

sample usage:

for r in Retrying(interval=self.waitTimeout):
    if self.fileExists():
        return True
    logging.debug('Waiting file %s attempt %d.', self.waitFile, r.attempt)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment