codekingpro/portable-devtools

1from __future__ import annotations
2
3import time
4
5
6__all__ = ["Deadline"]
7
8
9class Deadline:
10    """
11    Manage timeouts across multiple steps.
12
13    Args:
14        timeout: Time available in seconds or :obj:`None` if there is no limit.
15
16    """
17
18    def __init__(self, timeout: float | None) -> None:
19        self.deadline: float | None
20        if timeout is None:
21            self.deadline = None
22        else:
23            self.deadline = time.monotonic() + timeout
24
25    def timeout(self, *, raise_if_elapsed: bool = True) -> float | None:
26        """
27        Calculate a timeout from a deadline.
28
29        Args:
30            raise_if_elapsed: Whether to raise :exc:`TimeoutError`
31                if the deadline lapsed.
32
33        Raises:
34            TimeoutError: If the deadline lapsed.
35
36        Returns:
37            Time left in seconds or :obj:`None` if there is no limit.
38
39        """
40        if self.deadline is None:
41            return None
42        timeout = self.deadline - time.monotonic()
43        if raise_if_elapsed and timeout <= 0:
44            raise TimeoutError("timed out")
45        return timeout
46