lowlevel.py 6.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228
  1. from __future__ import annotations
  2. __all__ = (
  3. "EventLoopToken",
  4. "RunvarToken",
  5. "RunVar",
  6. "checkpoint",
  7. "checkpoint_if_cancelled",
  8. "cancel_shielded_checkpoint",
  9. "current_token",
  10. )
  11. import enum
  12. from dataclasses import dataclass
  13. from types import TracebackType
  14. from typing import TYPE_CHECKING, Any, Generic, Literal, TypeVar, final, overload
  15. from weakref import WeakKeyDictionary
  16. from ._core._eventloop import get_async_backend
  17. if TYPE_CHECKING:
  18. from .abc import AsyncBackend
  19. T = TypeVar("T")
  20. D = TypeVar("D")
  21. async def checkpoint() -> None:
  22. """
  23. Check for cancellation and allow the scheduler to switch to another task.
  24. Equivalent to (but more efficient than)::
  25. await checkpoint_if_cancelled()
  26. await cancel_shielded_checkpoint()
  27. .. versionadded:: 3.0
  28. """
  29. await get_async_backend().checkpoint()
  30. async def checkpoint_if_cancelled() -> None:
  31. """
  32. Enter a checkpoint if the enclosing cancel scope has been cancelled.
  33. This does not allow the scheduler to switch to a different task.
  34. .. versionadded:: 3.0
  35. """
  36. await get_async_backend().checkpoint_if_cancelled()
  37. async def cancel_shielded_checkpoint() -> None:
  38. """
  39. Allow the scheduler to switch to another task but without checking for cancellation.
  40. Equivalent to (but potentially more efficient than)::
  41. with CancelScope(shield=True):
  42. await checkpoint()
  43. .. versionadded:: 3.0
  44. """
  45. await get_async_backend().cancel_shielded_checkpoint()
  46. @final
  47. @dataclass(frozen=True, repr=False)
  48. class EventLoopToken:
  49. """
  50. An opaque object that holds a reference to an event loop.
  51. .. versionadded:: 4.11.0
  52. """
  53. backend_class: type[AsyncBackend]
  54. native_token: object
  55. def current_token() -> EventLoopToken:
  56. """
  57. Return a token object that can be used to call code in the current event loop from
  58. another thread.
  59. :raises NoEventLoopError: if no supported asynchronous event loop is running in the
  60. current thread
  61. .. versionadded:: 4.11.0
  62. """
  63. backend_class = get_async_backend()
  64. raw_token = backend_class.current_token()
  65. return EventLoopToken(backend_class, raw_token)
  66. _run_vars: WeakKeyDictionary[object, dict[RunVar[Any], Any]] = WeakKeyDictionary()
  67. class _NoValueSet(enum.Enum):
  68. NO_VALUE_SET = enum.auto()
  69. class RunvarToken(Generic[T]):
  70. """
  71. A token that can be used to restore a :class:`RunVar` to its previous value.
  72. Returned by :meth:`RunVar.set`. Can be used as a context manager to automatically
  73. reset the variable on exit, or passed directly to :meth:`RunVar.reset`.
  74. """
  75. __slots__ = "_var", "_value", "_redeemed"
  76. def __init__(self, var: RunVar[T], value: T | Literal[_NoValueSet.NO_VALUE_SET]):
  77. self._var = var
  78. self._value: T | Literal[_NoValueSet.NO_VALUE_SET] = value
  79. self._redeemed = False
  80. def __enter__(self) -> RunvarToken[T]:
  81. return self
  82. def __exit__(
  83. self,
  84. exc_type: type[BaseException] | None,
  85. exc_val: BaseException | None,
  86. exc_tb: TracebackType | None,
  87. ) -> None:
  88. self._var.reset(self)
  89. class RunVar(Generic[T]):
  90. """
  91. Like a :class:`~contextvars.ContextVar`, except scoped to the running event loop.
  92. Can be used as a context manager, Just like :class:`~contextvars.ContextVar`, that
  93. will reset the variable to its previous value when the context block is exited.
  94. """
  95. __slots__ = "_name", "_default"
  96. NO_VALUE_SET: Literal[_NoValueSet.NO_VALUE_SET] = _NoValueSet.NO_VALUE_SET
  97. def __init__(
  98. self, name: str, default: T | Literal[_NoValueSet.NO_VALUE_SET] = NO_VALUE_SET
  99. ):
  100. self._name = name
  101. self._default = default
  102. @property
  103. def _current_vars(self) -> dict[RunVar[T], T]:
  104. native_token = current_token().native_token
  105. try:
  106. return _run_vars[native_token]
  107. except KeyError:
  108. run_vars = _run_vars[native_token] = {}
  109. return run_vars
  110. @overload
  111. def get(self, default: D) -> T | D: ...
  112. @overload
  113. def get(self) -> T: ...
  114. def get(
  115. self, default: D | Literal[_NoValueSet.NO_VALUE_SET] = NO_VALUE_SET
  116. ) -> T | D:
  117. """
  118. Return the current value of this run variable.
  119. :param default: a fallback value to return if no value has been set
  120. :return: the current value, the provided default, or the variable's own default
  121. :raises LookupError: if no value is set and no default is available
  122. """
  123. try:
  124. return self._current_vars[self]
  125. except KeyError:
  126. if default is not RunVar.NO_VALUE_SET:
  127. return default
  128. elif self._default is not RunVar.NO_VALUE_SET:
  129. return self._default
  130. raise LookupError(
  131. f'Run variable "{self._name}" has no value and no default set'
  132. )
  133. def set(self, value: T) -> RunvarToken[T]:
  134. """
  135. Set the value of this run variable for the current event loop.
  136. :param value: the new value
  137. :return: a token that can be used to restore the previous value
  138. """
  139. current_vars = self._current_vars
  140. token = RunvarToken(self, current_vars.get(self, RunVar.NO_VALUE_SET))
  141. current_vars[self] = value
  142. return token
  143. def reset(self, token: RunvarToken[T]) -> None:
  144. """
  145. Restore this run variable to the value it held before the matching :meth:`set`.
  146. :param token: the token returned by :meth:`set`
  147. :raises ValueError: if the token belongs to a different :class:`RunVar` or the token
  148. has already been used
  149. """
  150. if token._var is not self:
  151. raise ValueError("This token does not belong to this RunVar")
  152. if token._redeemed:
  153. raise ValueError("This token has already been used")
  154. if token._value is _NoValueSet.NO_VALUE_SET:
  155. try:
  156. del self._current_vars[self]
  157. except KeyError:
  158. pass
  159. else:
  160. self._current_vars[self] = token._value
  161. token._redeemed = True
  162. def __repr__(self) -> str:
  163. return f"<RunVar name={self._name!r}>"