_eventloop.py 6.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240
  1. from __future__ import annotations
  2. import math
  3. import sys
  4. import threading
  5. from collections.abc import Awaitable, Callable, Generator
  6. from contextlib import contextmanager
  7. from contextvars import Token
  8. from importlib import import_module
  9. from typing import TYPE_CHECKING, Any, TypeVar
  10. from ._exceptions import NoEventLoopError
  11. if sys.version_info >= (3, 11):
  12. from typing import TypeVarTuple, Unpack
  13. else:
  14. from typing_extensions import TypeVarTuple, Unpack
  15. sniffio: Any
  16. try:
  17. import sniffio
  18. except ModuleNotFoundError:
  19. sniffio = None
  20. if TYPE_CHECKING:
  21. from ..abc import AsyncBackend
  22. # This must be updated when new backends are introduced
  23. BACKENDS = "asyncio", "trio"
  24. T_Retval = TypeVar("T_Retval")
  25. PosArgsT = TypeVarTuple("PosArgsT")
  26. threadlocals = threading.local()
  27. loaded_backends: dict[str, type[AsyncBackend]] = {}
  28. def run(
  29. func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval]],
  30. *args: Unpack[PosArgsT],
  31. backend: str = "asyncio",
  32. backend_options: dict[str, Any] | None = None,
  33. ) -> T_Retval:
  34. """
  35. Run the given coroutine function in an asynchronous event loop.
  36. The current thread must not be already running an event loop.
  37. :param func: a coroutine function
  38. :param args: positional arguments to ``func``
  39. :param backend: name of the asynchronous event loop implementation – currently
  40. either ``asyncio`` or ``trio``
  41. :param backend_options: keyword arguments to call the backend ``run()``
  42. implementation with (documented :ref:`here <backend options>`)
  43. :return: the return value of the coroutine function
  44. :raises RuntimeError: if an asynchronous event loop is already running in this
  45. thread
  46. :raises LookupError: if the named backend is not found
  47. """
  48. if asynclib_name := current_async_library():
  49. raise RuntimeError(f"Already running {asynclib_name} in this thread")
  50. try:
  51. async_backend = get_async_backend(backend)
  52. except ImportError as exc:
  53. if backend in BACKENDS:
  54. raise LookupError(
  55. f"Backend {backend!r} is not available. "
  56. f"Install it with: pip install anyio[{backend}]"
  57. ) from exc
  58. raise LookupError(f"No such backend: {backend}") from exc
  59. token = None
  60. if asynclib_name is None:
  61. # Since we're in control of the event loop, we can cache the name of the async
  62. # library
  63. token = set_current_async_library(backend)
  64. try:
  65. backend_options = backend_options or {}
  66. return async_backend.run(func, args, {}, backend_options)
  67. finally:
  68. reset_current_async_library(token)
  69. async def sleep(delay: float) -> None:
  70. """
  71. Pause the current task for the specified duration.
  72. :param delay: the duration, in seconds
  73. """
  74. return await get_async_backend().sleep(delay)
  75. async def sleep_forever() -> None:
  76. """
  77. Pause the current task until it's cancelled.
  78. This is a shortcut for ``sleep(math.inf)``.
  79. .. versionadded:: 3.1
  80. """
  81. await sleep(math.inf)
  82. async def sleep_until(deadline: float) -> None:
  83. """
  84. Pause the current task until the given time.
  85. :param deadline: the absolute time to wake up at (according to the internal
  86. monotonic clock of the event loop)
  87. .. versionadded:: 3.1
  88. """
  89. now = current_time()
  90. await sleep(max(deadline - now, 0))
  91. def current_time() -> float:
  92. """
  93. Return the current value of the event loop's internal clock.
  94. :return: the clock value (seconds)
  95. :raises NoEventLoopError: if no supported asynchronous event loop is running in the
  96. current thread
  97. """
  98. return get_async_backend().current_time()
  99. def get_all_backends() -> tuple[str, ...]:
  100. """Return a tuple of the names of all built-in backends."""
  101. return BACKENDS
  102. def get_available_backends() -> tuple[str, ...]:
  103. """
  104. Test for the availability of built-in backends.
  105. :return a tuple of the built-in backend names that were successfully imported
  106. .. versionadded:: 4.12
  107. """
  108. available_backends: list[str] = []
  109. for backend_name in get_all_backends():
  110. try:
  111. get_async_backend(backend_name)
  112. except ImportError:
  113. continue
  114. available_backends.append(backend_name)
  115. return tuple(available_backends)
  116. def get_cancelled_exc_class() -> type[BaseException]:
  117. """
  118. Return the current async library's cancellation exception class.
  119. :raises NoEventLoopError: if no supported asynchronous event loop is running in the
  120. current thread
  121. """
  122. return get_async_backend().cancelled_exception_class()
  123. #
  124. # Private API
  125. #
  126. @contextmanager
  127. def claim_worker_thread(
  128. backend_class: type[AsyncBackend], token: object
  129. ) -> Generator[Any, None, None]:
  130. from ..lowlevel import EventLoopToken
  131. threadlocals.current_token = EventLoopToken(backend_class, token)
  132. try:
  133. yield
  134. finally:
  135. del threadlocals.current_token
  136. def get_async_backend(asynclib_name: str | None = None) -> type[AsyncBackend]:
  137. if asynclib_name is None:
  138. asynclib_name = current_async_library()
  139. if not asynclib_name:
  140. raise NoEventLoopError(
  141. f"Not currently running on any asynchronous event loop. "
  142. f"Available async backends: {', '.join(get_all_backends())}"
  143. )
  144. # We use our own dict instead of sys.modules to get the already imported back-end
  145. # class because the appropriate modules in sys.modules could potentially be only
  146. # partially initialized
  147. try:
  148. return loaded_backends[asynclib_name]
  149. except KeyError:
  150. module = import_module(f"anyio._backends._{asynclib_name}")
  151. loaded_backends[asynclib_name] = module.backend_class
  152. return module.backend_class
  153. def current_async_library() -> str | None:
  154. if sniffio is None:
  155. # If sniffio is not installed, we assume we're either running asyncio or nothing
  156. import asyncio
  157. try:
  158. asyncio.get_running_loop()
  159. return "asyncio"
  160. except RuntimeError:
  161. pass
  162. else:
  163. try:
  164. return sniffio.current_async_library()
  165. except sniffio.AsyncLibraryNotFoundError:
  166. pass
  167. return None
  168. def set_current_async_library(asynclib_name: str | None) -> Token | None:
  169. # no-op if sniffio is not installed
  170. if sniffio is None:
  171. return None
  172. return sniffio.current_async_library_cvar.set(asynclib_name)
  173. def reset_current_async_library(token: Token | None) -> None:
  174. if token is not None:
  175. sniffio.current_async_library_cvar.reset(token)