Source code for stdlibx.cancel._types

 1from __future__ import annotations
 2
 3from typing import TYPE_CHECKING, Literal, NoReturn, Protocol
 4
 5if TYPE_CHECKING:
 6    from collections.abc import Callable
 7
 8

[docs]
 9class CancellationToken(Protocol):
10    """A cooperative cancellation token interface. Represents a handle that can
11    be checked or awaited to respond to cancellation events triggered elsewhere
12    """
13

[docs]
14    def register(self, fn: Callable[[Exception], None]) -> None:
15        """Register a callback to be invoked when the token is cancelled.
16
17        :param fn: A function to call when cancellation occurs. Receives a
18            `Exception`.
19        """
20        ...

21

[docs]
22    def is_cancelled(self) -> bool:
23        """Check whether the token has been cancelled.
24
25        :returns: True if the token has been cancelled, False otherwise.
26        """
27        ...

28

[docs]
29    def get_error(self) -> Exception | None:
30        """Get the cancellation error if the token has been cancelled.
31
32        :returns: The `Exception` if cancelled, or None.
33        """
34        ...

35

[docs]
36    def raise_if_cancelled(self):
37        """Raise the cancellation error if the token has been cancelled.
38
39        :raises Exception: If the token is cancelled.
40        """
41        ...

42

[docs]
43    def wait(self, timeout: float | None) -> Exception | None:
44        """Block until the token is cancelled or the timeout expires.
45
46        :param timeout: Maximum time to wait in seconds, or None to wait
47            indefinitely.
48        :returns: The `Exception` if cancelled, or None if the timeout expires.
49        """
50        ...


51
52

[docs]
53class CancelledToken(CancellationToken, Protocol):
54    """A token that is already cancelled. This specialization of `Token`
55    ensures that cancellation has occurred and provides non-optional return
56    types
57
58    Example:
59
60    .. code-block:: python
61        :linenos:
62
63        if is_token_cancelled(token):
64            reveal_type(token)  # CancelledToken
65    """
66

[docs]
67    def is_cancelled(self) -> Literal[True]:
68        """Always returns True, since the token is cancelled.
69
70        :returns: True
71        """
72        ...

73

[docs]
74    def get_error(self) -> Exception:
75        """Get the cancellation error.
76
77        :returns: The `Exception` associated with this cancelled token.
78        """
79        ...

80

[docs]
81    def raise_if_cancelled(self) -> NoReturn:
82        """Always raises the cancellation error.
83
84        :raises Exception: The associated cancellation error.
85        """
86        ...

87

[docs]
88    def wait(self, timeout: float | None) -> Exception:
89        """Immediately returns the cancellation error.
90
91        :param timeout: Ignored. Present for interface compatibility.
92        :returns: The `Exception` associated with this cancelled token.
93        """
94        ...