Source code for bok_choy.promise

"""
Variation on the "promise" design pattern.
Promises make it easier to handle asynchronous operations correctly.
"""

import time
import logging


LOGGER = logging.getLogger(__name__)


[docs]class BrokenPromise(Exception): """ The promise was not satisfied within the time constraints. """ def __init__(self, promise): """ Configure the broken promise error. Args: promise (Promise): The promise that was not satisfied. """ super().__init__() self._promise = promise def __str__(self): return f"Promise not satisfied: {self._promise}"
[docs]class Promise: """ Check that an asynchronous action completed, blocking until it does or timeout / try limits are reached. """ # pylint: disable=too-many-arguments def __init__(self, check_func, description, try_limit=None, try_interval=0.5, timeout=30): """ Configure the `Promise`. The `Promise` will poll `check_func()` until either: * The promise is satisfied * The promise runs out of tries (checks more than `try_limit` times) * The promise runs out of time (takes longer than `timeout` seconds) If the try_limit or timeout is reached without success, then the promise is "broken" and an exception will be raised. Note that if you specify a try_limit but not a timeout, the default timeout is still used. This is to prevent an inadvertent infinite loop. If you want to make sure that the try_limit expires first (and thus that many attempts will be made), then you should also pass in a larger value for timeout. `description` is a string that will be included in the exception to make debugging easier. Example: .. code:: python # Dummy check function that indicates the promise is always satisfied check_func = lambda: (True, "Hello world!") # Check up to 5 times if the operation has completed result = Promise(check_func, "Operation has completed", try_limit=5).fulfill() Args: check_func (callable): A function that accepts no arguments and returns a `(is_satisfied, result)` tuple, where `is_satisfied` is a boolean indiating whether the promise was satisfied, and `result` is a value to return from the fulfilled `Promise`. description (str): Description of the `Promise`, used in log messages. Keyword Args: try_limit (int or None): Number of attempts to make to satisfy the `Promise`. Can be `None` to disable the limit. try_interval (float): Number of seconds to wait between attempts. timeout (float): Maximum number of seconds to wait for the `Promise` to be satisfied before timing out. Returns: Promise """ self._check_func = check_func self._description = description self._try_limit = try_limit self._try_interval = try_interval self._timeout = timeout self._num_tries = 0
[docs] def fulfill(self): """ Evaluate the promise and return the result. Returns: The result of the `Promise` (second return value from the `check_func`) Raises: BrokenPromise: the `Promise` was not satisfied within the time or attempt limits. """ is_fulfilled, result = self._check_fulfilled() if is_fulfilled: return result raise BrokenPromise(self)
def __str__(self): return str(self._description) def _check_fulfilled(self): """ Return tuple `(is_fulfilled, result)` where `is_fulfilled` is a boolean indicating whether the promise has been fulfilled and `result` is the value to pass to the `with` block. """ is_fulfilled = False result = None start_time = time.time() # Check whether the promise has been fulfilled until we run out of time or attempts while self._has_time_left(start_time) and self._has_more_tries(): # Keep track of how many attempts we've made so far self._num_tries += 1 is_fulfilled, result = self._check_func() # If the promise is satisfied, then continue execution if is_fulfilled: break # Delay between checks time.sleep(self._try_interval) return is_fulfilled, result def _has_time_left(self, start_time): """ Return True if the elapsed time is less than the timeout. """ return time.time() - start_time < self._timeout def _has_more_tries(self): """ Return True if the promise has additional tries. If `_try_limit` is `None`, always return True. """ if self._try_limit is None: return True return self._num_tries < self._try_limit
[docs]class EmptyPromise(Promise): # pylint: disable=too-few-public-methods """ A promise that has no result value. """ def __init__(self, check_func, description, **kwargs): """ Configure the promise. Unlike a regular `Promise`, the `check_func()` does NOT return a tuple with a result value. That's why the promise is "empty" -- you don't get anything back. Example usage: .. code:: python # This will block until `is_done` returns `True` or we reach the timeout limit. EmptyPromise(lambda: is_done('test'), "Test operation is done").fulfill() Args: check_func (callable): Function that accepts no arguments and returns a boolean indicating whether the promise is fulfilled. description (str): Description of the Promise, used in log messages. Returns: EmptyPromise """ full_check_func = lambda: (check_func(), None) # pylint: disable=unnecessary-lambda-assignment super().__init__(full_check_func, description, **kwargs)