future

dandy.core.future.future

async_executor = concurrent.futures.ThreadPoolExecutor() module-attribute

FutureResultType = TypeVar('FutureResultType') module-attribute

AsyncFuture

Bases: Generic[FutureResultType]

Source code in dandy/core/future/future.py

def __init__(self, callable_: Callable[..., FutureResultType], *args, **kwargs):
    self._future: Future = async_executor.submit(callable_, *args, **kwargs)
    self._future_start_time = perf_counter()

    self._result = None
    self._result_timeout = None
    self._using_result_timeout = False

result property

cancel

Source code in dandy/core/future/future.py

def cancel(self) -> bool:
    return self._future.cancel()

set_timeout

Source code in dandy/core/future/future.py

def set_timeout(self, seconds: int):
    self._using_result_timeout = True
    self._result_timeout = seconds