Orion Client Utilities¶
Backoff(cap: float = 30.0, base: float = 0.1, backoff_dict: Union[Dict[int, Tuple[int, int]], NoneType] = None)¶
Utility for Jittered Backoff
Instantiate and call sleep() as many times as necessary
Uses the decorrelated jitter backoff mechanism described here: https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/
- @param (cap) default=30.0: default maximum sleep in seconds @param (base) default=0.1: default initial sleep in seconds @param (backoff_dict) default=None: f: status_code –> (base, cap)
If backoff_dict does not contain a key for a status_code, then the default cap and base are used.
from requests import session from orionclient.utils import Backoff def retry_get_with_backoff(url, attempts=10, request_timeout=10): sess = session() tries = 0 sleeper = Backoff() resp = None while tries < attempts: resp = sess.get(url, timeout=request_timeout) if not resp.ok: tries += 1 sleeper.sleep() continue return resp return resp
Exponential sleep based on the number of tries
TemporaryPath(suffix='', prefix='tmp', dir=None, delete=True)¶
Provides a temporary file path, suggested usage is as a context manager.
This is safe to use cross-platform where NamedTemporaryFile is not.
with TemporaryPath(suffix="bar") as path: with open(path, "wb") as ofs: ofs.write(b"bytes")