wip

Author: jooola

hcloud/_utils.py

@@ -1,10 +1,9 @@
 from __future__ import annotations
 
+import time
 from collections.abc import Iterable, Iterator
 from itertools import islice
-from typing import TypeVar
-from functools import wraps
-from concurrent.futures import ThreadPoolExecutor
+from typing import Callable, TypeVar
 
 T = TypeVar("T")
 
@@ -21,16 +20,41 @@ def batched(iterable: Iterable[T], size: int) -> Iterator[tuple[T, ...]]:
         yield batch
 
 
-def with_timeout(seconds: float | None):
-    def decorator(f):
-        @wraps(f)
-        def wrapper(*args, **kwargs):
-            if seconds:
-                with ThreadPoolExecutor(max_workers=1) as executor:
-                    future = executor.submit(f, *args, **kwargs)
-                    return future.result(timeout=seconds)
-            else:
-                return f(*args, **kwargs)
+def waiter(timeout: float | None = None) -> Callable[[float], bool]:
+    """
+    Waiter returns a wait function that sleeps the specified amount of seconds, and
+    handles timeouts.
+
+    The wait function returns True if the timeout was reached, False otherwise.
+
+    :param timeout: Timeout in seconds, defaults to None.
+    :return: Wait function.
+    """
+
+    if timeout:
+        deadline = time.time() + timeout
+
+        def wait(seconds: float) -> bool:
+            now = time.time()
+
+            # Timeout if the deadline exceeded.
+            if deadline < now:
+                return True
+
+            # The deadline is not exceeded after the sleep time.
+            if now + seconds < deadline:
+                time.sleep(seconds)
+                return False
+
+            # The deadline is exceeded after the sleep time, clamp sleep time to
+            # deadline, and allow one last attempt until next wait call.
+            time.sleep(deadline - now)
+            return False
+
+    else:
+
+        def wait(seconds: float) -> bool:
+            time.sleep(seconds)
+            return False
 
-        return wrapper
-    return decorator
+    return wait

hcloud/actions/client.py

@@ -1,12 +1,9 @@
 from __future__ import annotations
 
-import time
 import warnings
-from concurrent.futures import ThreadPoolExecutor
-from threading import Event
 from typing import TYPE_CHECKING, Any, Callable, NamedTuple
 
-from .._utils import batched, with_timeout
+from .._utils import batched, waiter
 from ..core import BoundModelBase, ClientEntityBase, Meta
 from .domain import Action, ActionFailedException, ActionTimeoutException
 
@@ -19,18 +16,24 @@ class BoundAction(BoundModelBase, Action):
 
     model = Action
 
-    def wait_until_finished(self, max_retries: int | None = None) -> None:
+    def wait_until_finished(
+        self,
+        max_retries: int | None = None,
+        *,
+        timeout: float | None = None,
+    ) -> None:
         """Wait until the specific action has status=finished.
 
         :param max_retries: int Specify how many retries will be performed before an ActionTimeoutException will be raised.
         :raises: ActionFailedException when action is finished with status==error
-        :raises: ActionTimeoutException when Action is still in status==running after max_retries is reached.
+        :raises: ActionTimeoutException when Action is still in status==running after max_retries or timeout is reached.
         """
         if max_retries is None:
             # pylint: disable=protected-access
             max_retries = self._client._client._poll_max_retries
 
         retries = 0
+        wait = waiter(timeout)
         while True:
             self.reload()
             if self.status != Action.STATUS_RUNNING:
@@ -39,8 +42,8 @@ def wait_until_finished(self, max_retries: int | None = None) -> None:
             retries += 1
             if retries < max_retries:
                 # pylint: disable=protected-access
-                time.sleep(self._client._client._poll_interval_func(retries))
-                continue
+                if not wait(self._client._client._poll_interval_func(retries)):
+                    continue
 
             raise ActionTimeoutException(action=self)
 
@@ -170,9 +173,10 @@ def _get_list_by_ids(self, ids: list[int]) -> list[BoundAction]:
 
     def wait_for_function(
         self,
-        stop: Event,
         handle_update: Callable[[BoundAction], None],
         actions: list[Action | BoundAction],
+        *,
+        timeout: float | None = None,
     ) -> list[BoundAction]:
         """
         Waits until all Actions succeed by polling the API at the interval defined by
@@ -183,6 +187,7 @@ def wait_for_function(
 
         :param handle_update: Function called every time an Action is updated.
         :param actions: List of Actions to wait for.
+        :param timeout: Timeout in seconds.
         :raises: ActionFailedException when an Action failed.
         :return: List of succeeded Actions.
         """
@@ -191,10 +196,12 @@ def wait_for_function(
         completed: list[BoundAction] = []
 
         retries = 0
+        wait = waiter(timeout)
         while len(running_ids):
             # pylint: disable=protected-access
-            if stop.wait(self._client._poll_interval_func(retries)):
-                raise ActionTimeoutException()
+            if wait(self._client._poll_interval_func(retries)):
+                # TODO: How to raise a timeout exception for many actiosn without exception group.
+                raise ActionTimeoutException("")
 
             retries += 1
 
@@ -212,6 +219,7 @@ def wait_for_function(
     def wait_for(
         self,
         actions: list[Action | BoundAction],
+        *,
         timeout: float | None = None,
     ) -> list[BoundAction]:
         """
@@ -222,6 +230,7 @@ def wait_for(
         If a single Action fails, the function will stop waiting and raise ActionFailedException.
 
         :param actions: List of Actions to wait for.
+        :param timeout: Timeout in seconds.
         :raises: ActionFailedException when an Action failed.
         :raises: TimeoutError when the Actions did not succeed before timeout.
         :return: List of succeeded Actions.
@@ -231,10 +240,7 @@ def handle_update(update: BoundAction) -> None:
             if update.status == Action.STATUS_ERROR:
                 raise ActionFailedException(action=update)
 
-        def run() -> list[BoundAction]:
-            return self.wait_for_function(handle_update, actions)
-
-        return with_timeout(timeout)(run)()
+        return self.wait_for_function(handle_update, actions, timeout=timeout)
 
     def get_list(
         self,

tests/unit/test_utils.py

@@ -1,7 +1,21 @@
 from __future__ import annotations
 
-from hcloud._utils import batched
+import time
+
+from hcloud._utils import batched, waiter
 
 
 def test_batched():
     assert list(o for o in batched([1, 2, 3, 4, 5], 2)) == [(1, 2), (3, 4), (5,)]
+
+
+def test_waiter():
+    wait = waiter(timeout=0.2)
+    assert wait(0.1) is False
+    time.sleep(0.2)
+    assert wait(1) is True
+
+    # Clamp sleep to deadline
+    wait = waiter(timeout=0.2)
+    assert wait(0.3) is False
+    assert wait(1) is True