codekingpro/portable-devtools

1from __future__ import annotations2 3import math4import sys5import threading6from collections.abc import Awaitable, Callable, Generator7from contextlib import contextmanager8from contextvars import Token9from importlib import import_module10from typing import TYPE_CHECKING, Any, TypeVar11 12from ._exceptions import NoEventLoopError13 14if sys.version_info >= (3, 11):15    from typing import TypeVarTuple, Unpack16else:17    from typing_extensions import TypeVarTuple, Unpack18 19sniffio: Any20try:21    import sniffio22except ModuleNotFoundError:23    sniffio = None24 25if TYPE_CHECKING:26    from ..abc import AsyncBackend27 28# This must be updated when new backends are introduced29BACKENDS = "asyncio", "trio"30 31T_Retval = TypeVar("T_Retval")32PosArgsT = TypeVarTuple("PosArgsT")33 34threadlocals = threading.local()35loaded_backends: dict[str, type[AsyncBackend]] = {}36 37 38def run(39    func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval]],40    *args: Unpack[PosArgsT],41    backend: str = "asyncio",42    backend_options: dict[str, Any] | None = None,43) -> T_Retval:44    """45    Run the given coroutine function in an asynchronous event loop.46 47    The current thread must not be already running an event loop.48 49    :param func: a coroutine function50    :param args: positional arguments to ``func``51    :param backend: name of the asynchronous event loop implementation – currently52        either ``asyncio`` or ``trio``53    :param backend_options: keyword arguments to call the backend ``run()``54        implementation with (documented :ref:`here <backend options>`)55    :return: the return value of the coroutine function56    :raises RuntimeError: if an asynchronous event loop is already running in this57        thread58    :raises LookupError: if the named backend is not found59 60    """61    if asynclib_name := current_async_library():62        raise RuntimeError(f"Already running {asynclib_name} in this thread")63 64    try:65        async_backend = get_async_backend(backend)66    except ImportError as exc:67        raise LookupError(f"No such backend: {backend}") from exc68 69    token = None70    if asynclib_name is None:71        # Since we're in control of the event loop, we can cache the name of the async72        # library73        token = set_current_async_library(backend)74 75    try:76        backend_options = backend_options or {}77        return async_backend.run(func, args, {}, backend_options)78    finally:79        reset_current_async_library(token)80 81 82async def sleep(delay: float) -> None:83    """84    Pause the current task for the specified duration.85 86    :param delay: the duration, in seconds87 88    """89    return await get_async_backend().sleep(delay)90 91 92async def sleep_forever() -> None:93    """94    Pause the current task until it's cancelled.95 96    This is a shortcut for ``sleep(math.inf)``.97 98    .. versionadded:: 3.199 100    """101    await sleep(math.inf)102 103 104async def sleep_until(deadline: float) -> None:105    """106    Pause the current task until the given time.107 108    :param deadline: the absolute time to wake up at (according to the internal109        monotonic clock of the event loop)110 111    .. versionadded:: 3.1112 113    """114    now = current_time()115    await sleep(max(deadline - now, 0))116 117 118def current_time() -> float:119    """120    Return the current value of the event loop's internal clock.121 122    :return: the clock value (seconds)123    :raises NoEventLoopError: if no supported asynchronous event loop is running in the124        current thread125 126    """127    return get_async_backend().current_time()128 129 130def get_all_backends() -> tuple[str, ...]:131    """Return a tuple of the names of all built-in backends."""132    return BACKENDS133 134 135def get_available_backends() -> tuple[str, ...]:136    """137    Test for the availability of built-in backends.138 139    :return a tuple of the built-in backend names that were successfully imported140 141    .. versionadded:: 4.12142 143    """144    available_backends: list[str] = []145    for backend_name in get_all_backends():146        try:147            get_async_backend(backend_name)148        except ImportError:149            continue150 151        available_backends.append(backend_name)152 153    return tuple(available_backends)154 155 156def get_cancelled_exc_class() -> type[BaseException]:157    """158    Return the current async library's cancellation exception class.159 160    :raises NoEventLoopError: if no supported asynchronous event loop is running in the161        current thread162 163    """164    return get_async_backend().cancelled_exception_class()165 166 167#168# Private API169#170 171 172@contextmanager173def claim_worker_thread(174    backend_class: type[AsyncBackend], token: object175) -> Generator[Any, None, None]:176    from ..lowlevel import EventLoopToken177 178    threadlocals.current_token = EventLoopToken(backend_class, token)179    try:180        yield181    finally:182        del threadlocals.current_token183 184 185def get_async_backend(asynclib_name: str | None = None) -> type[AsyncBackend]:186    if asynclib_name is None:187        asynclib_name = current_async_library()188        if not asynclib_name:189            raise NoEventLoopError(190                f"Not currently running on any asynchronous event loop. "191                f"Available async backends: {', '.join(get_all_backends())}"192            )193 194    # We use our own dict instead of sys.modules to get the already imported back-end195    # class because the appropriate modules in sys.modules could potentially be only196    # partially initialized197    try:198        return loaded_backends[asynclib_name]199    except KeyError:200        module = import_module(f"anyio._backends._{asynclib_name}")201        loaded_backends[asynclib_name] = module.backend_class202        return module.backend_class203 204 205def current_async_library() -> str | None:206    if sniffio is None:207        # If sniffio is not installed, we assume we're either running asyncio or nothing208        import asyncio209 210        try:211            asyncio.get_running_loop()212            return "asyncio"213        except RuntimeError:214            pass215    else:216        try:217            return sniffio.current_async_library()218        except sniffio.AsyncLibraryNotFoundError:219            pass220 221    return None222 223 224def set_current_async_library(asynclib_name: str | None) -> Token | None:225    # no-op if sniffio is not installed226    if sniffio is None:227        return None228 229    return sniffio.current_async_library_cvar.set(asynclib_name)230 231 232def reset_current_async_library(token: Token | None) -> None:233    if token is not None:234        sniffio.current_async_library_cvar.reset(token)235