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