Skip to content

future

dandy.core.future.future

thread_pool_executor = concurrent.futures.ThreadPoolExecutor(max_workers=(settings.FUTURES_MAX_WORKERS)) module-attribute

R = TypeVar('R') module-attribute

AsyncFuture

Bases: Generic[R]

Source code in dandy/core/future/future.py 
28
29
30
31
32
33
34
def __init__(self, callable_: Callable[..., R], *args, **kwargs):
    self._future: Future = thread_pool_executor.submit(callable_, *args, **kwargs)

    self._result: R = None
    self._result_fetched: bool = False
    self._result_timeout: float | None = None
    self._lock = threading.RLock()

result property

cancel

Source code in dandy/core/future/future.py 
36
37
def cancel(self) -> bool:
    return self._future.cancel()

cancelled

Source code in dandy/core/future/future.py 
39
40
def cancelled(self) -> bool:
    return self._future.cancelled()

done

Source code in dandy/core/future/future.py 
42
43
def done(self) -> bool:
    return self._future.done()

get_result

Source code in dandy/core/future/future.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def get_result(self, timeout_seconds: float | None = None) -> R:
    with self._lock:
        if self._result_fetched:
            return self._result

        try:
            self.set_timeout(timeout_seconds)

            self._result: R = self._future.result(timeout=self._result_timeout)
            self._result_fetched = True

        except concurrent.futures.TimeoutError as error:
            self.cancel()
            message = f"Future timed out after {self._result_timeout} seconds"
            raise FutureRecoverableException(message) from error

        return self._result

set_timeout

Source code in dandy/core/future/future.py 
67
68
69
70
71
72
def set_timeout(self, seconds: float | None = None):
    if seconds is not None and seconds <= 0:
        message = f"Future timeout must be greater than 0.0, not {seconds}"
        raise FutureCriticalException(message)

    self._result_timeout = seconds