1
1
import enum
2
2
3
3
from types import TracebackType
4
- from typing import final , Optional , Type
5
4
6
5
from . import events
7
6
from . import exceptions
@@ -23,14 +22,13 @@ class _State(enum.Enum):
23
22
EXITED = "finished"
24
23
25
24
26
- @final
27
25
class Timeout :
28
26
"""Asynchronous context manager for cancelling overdue coroutines.
29
27
30
28
Use `timeout()` or `timeout_at()` rather than instantiating this class directly.
31
29
"""
32
30
33
- def __init__ (self , when : Optional [ float ] ) -> None :
31
+ def __init__ (self , when : float | None ) -> None :
34
32
"""Schedule a timeout that will trigger at a given loop time.
35
33
36
34
- If `when` is `None`, the timeout will never trigger.
@@ -39,15 +37,15 @@ def __init__(self, when: Optional[float]) -> None:
39
37
"""
40
38
self ._state = _State .CREATED
41
39
42
- self ._timeout_handler : Optional [ events .TimerHandle ] = None
43
- self ._task : Optional [ tasks .Task ] = None
40
+ self ._timeout_handler : events .TimerHandle | None = None
41
+ self ._task : tasks .Task | None = None
44
42
self ._when = when
45
43
46
- def when (self ) -> Optional [ float ] :
44
+ def when (self ) -> float | None :
47
45
"""Return the current deadline."""
48
46
return self ._when
49
47
50
- def reschedule (self , when : Optional [ float ] ) -> None :
48
+ def reschedule (self , when : float | None ) -> None :
51
49
"""Reschedule the timeout."""
52
50
if self ._state is not _State .ENTERED :
53
51
if self ._state is _State .CREATED :
@@ -96,10 +94,10 @@ async def __aenter__(self) -> "Timeout":
96
94
97
95
async def __aexit__ (
98
96
self ,
99
- exc_type : Optional [ Type [ BaseException ]] ,
100
- exc_val : Optional [ BaseException ] ,
101
- exc_tb : Optional [ TracebackType ] ,
102
- ) -> Optional [ bool ] :
97
+ exc_type : type [ BaseException ] | None ,
98
+ exc_val : BaseException | None ,
99
+ exc_tb : TracebackType | None ,
100
+ ) -> bool | None :
103
101
assert self ._state in (_State .ENTERED , _State .EXPIRING )
104
102
105
103
if self ._timeout_handler is not None :
@@ -142,7 +140,7 @@ def _insert_timeout_error(exc_val: BaseException) -> None:
142
140
exc_val = exc_val .__context__
143
141
144
142
145
- def timeout (delay : Optional [ float ] ) -> Timeout :
143
+ def timeout (delay : float | None ) -> Timeout :
146
144
"""Timeout async context manager.
147
145
148
146
Useful in cases when you want to apply timeout logic around block
@@ -162,7 +160,7 @@ def timeout(delay: Optional[float]) -> Timeout:
162
160
return Timeout (loop .time () + delay if delay is not None else None )
163
161
164
162
165
- def timeout_at (when : Optional [ float ] ) -> Timeout :
163
+ def timeout_at (when : float | None ) -> Timeout :
166
164
"""Schedule the timeout at absolute time.
167
165
168
166
Like timeout() but argument gives absolute time in the same clock system
0 commit comments