Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/anyio/lowlevel.py: 49%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3__all__ = (
4 "EventLoopToken",
5 "RunvarToken",
6 "RunVar",
7 "checkpoint",
8 "checkpoint_if_cancelled",
9 "cancel_shielded_checkpoint",
10 "current_token",
11)
13import enum
14from dataclasses import dataclass
15from types import TracebackType
16from typing import TYPE_CHECKING, Any, Generic, Literal, TypeVar, final, overload
17from weakref import WeakKeyDictionary
19from ._core._eventloop import get_async_backend
21if TYPE_CHECKING:
22 from .abc import AsyncBackend
24T = TypeVar("T")
25D = TypeVar("D")
28async def checkpoint() -> None:
29 """
30 Check for cancellation and allow the scheduler to switch to another task.
32 Equivalent to (but more efficient than)::
34 await checkpoint_if_cancelled()
35 await cancel_shielded_checkpoint()
37 .. versionadded:: 3.0
39 """
40 await get_async_backend().checkpoint()
43async def checkpoint_if_cancelled() -> None:
44 """
45 Enter a checkpoint if the enclosing cancel scope has been cancelled.
47 This does not allow the scheduler to switch to a different task.
49 .. versionadded:: 3.0
51 """
52 await get_async_backend().checkpoint_if_cancelled()
55async def cancel_shielded_checkpoint() -> None:
56 """
57 Allow the scheduler to switch to another task but without checking for cancellation.
59 Equivalent to (but potentially more efficient than)::
61 with CancelScope(shield=True):
62 await checkpoint()
64 .. versionadded:: 3.0
66 """
67 await get_async_backend().cancel_shielded_checkpoint()
70@final
71@dataclass(frozen=True, repr=False)
72class EventLoopToken:
73 """
74 An opaque object that holds a reference to an event loop.
76 .. versionadded:: 4.11.0
77 """
79 backend_class: type[AsyncBackend]
80 native_token: object
83def current_token() -> EventLoopToken:
84 """
85 Return a token object that can be used to call code in the current event loop from
86 another thread.
88 :raises NoEventLoopError: if no supported asynchronous event loop is running in the
89 current thread
91 .. versionadded:: 4.11.0
93 """
94 backend_class = get_async_backend()
95 raw_token = backend_class.current_token()
96 return EventLoopToken(backend_class, raw_token)
99_run_vars: WeakKeyDictionary[object, dict[RunVar[Any], Any]] = WeakKeyDictionary()
102class _NoValueSet(enum.Enum):
103 NO_VALUE_SET = enum.auto()
106class RunvarToken(Generic[T]):
107 """
108 A token that can be used to restore a :class:`RunVar` to its previous value.
110 Returned by :meth:`RunVar.set`. Can be used as a context manager to automatically
111 reset the variable on exit, or passed directly to :meth:`RunVar.reset`.
112 """
114 __slots__ = "_var", "_value", "_redeemed"
116 def __init__(self, var: RunVar[T], value: T | Literal[_NoValueSet.NO_VALUE_SET]):
117 self._var = var
118 self._value: T | Literal[_NoValueSet.NO_VALUE_SET] = value
119 self._redeemed = False
121 def __enter__(self) -> RunvarToken[T]:
122 return self
124 def __exit__(
125 self,
126 exc_type: type[BaseException] | None,
127 exc_val: BaseException | None,
128 exc_tb: TracebackType | None,
129 ) -> None:
130 self._var.reset(self)
133class RunVar(Generic[T]):
134 """
135 Like a :class:`~contextvars.ContextVar`, except scoped to the running event loop.
137 Can be used as a context manager, Just like :class:`~contextvars.ContextVar`, that
138 will reset the variable to its previous value when the context block is exited.
139 """
141 __slots__ = "_name", "_default"
143 NO_VALUE_SET: Literal[_NoValueSet.NO_VALUE_SET] = _NoValueSet.NO_VALUE_SET
145 def __init__(
146 self, name: str, default: T | Literal[_NoValueSet.NO_VALUE_SET] = NO_VALUE_SET
147 ):
148 self._name = name
149 self._default = default
151 @property
152 def _current_vars(self) -> dict[RunVar[T], T]:
153 native_token = current_token().native_token
154 try:
155 return _run_vars[native_token]
156 except KeyError:
157 run_vars = _run_vars[native_token] = {}
158 return run_vars
160 @overload
161 def get(self, default: D) -> T | D: ...
163 @overload
164 def get(self) -> T: ...
166 def get(
167 self, default: D | Literal[_NoValueSet.NO_VALUE_SET] = NO_VALUE_SET
168 ) -> T | D:
169 """
170 Return the current value of this run variable.
172 :param default: a fallback value to return if no value has been set
173 :return: the current value, the provided default, or the variable's own default
174 :raises LookupError: if no value is set and no default is available
176 """
177 try:
178 return self._current_vars[self]
179 except KeyError:
180 if default is not RunVar.NO_VALUE_SET:
181 return default
182 elif self._default is not RunVar.NO_VALUE_SET:
183 return self._default
185 raise LookupError(
186 f'Run variable "{self._name}" has no value and no default set'
187 )
189 def set(self, value: T) -> RunvarToken[T]:
190 """
191 Set the value of this run variable for the current event loop.
193 :param value: the new value
194 :return: a token that can be used to restore the previous value
196 """
197 current_vars = self._current_vars
198 token = RunvarToken(self, current_vars.get(self, RunVar.NO_VALUE_SET))
199 current_vars[self] = value
200 return token
202 def reset(self, token: RunvarToken[T]) -> None:
203 """
204 Restore this run variable to the value it held before the matching :meth:`set`.
206 :param token: the token returned by :meth:`set`
207 :raises ValueError: if the token belongs to a different :class:`RunVar` or the token
208 has already been used
210 """
211 if token._var is not self:
212 raise ValueError("This token does not belong to this RunVar")
214 if token._redeemed:
215 raise ValueError("This token has already been used")
217 if token._value is _NoValueSet.NO_VALUE_SET:
218 try:
219 del self._current_vars[self]
220 except KeyError:
221 pass
222 else:
223 self._current_vars[self] = token._value
225 token._redeemed = True
227 def __repr__(self) -> str:
228 return f"<RunVar name={self._name!r}>"