codekingpro/portable-devtools

1from __future__ import annotations2 3__all__ = (4    "run_sync",5    "current_default_thread_limiter",6)7 8import sys9from collections.abc import Callable10from typing import TypeVar11from warnings import warn12 13from ._core._eventloop import get_async_backend14from .abc import CapacityLimiter15 16if sys.version_info >= (3, 11):17    from typing import TypeVarTuple, Unpack18else:19    from typing_extensions import TypeVarTuple, Unpack20 21T_Retval = TypeVar("T_Retval")22PosArgsT = TypeVarTuple("PosArgsT")23 24 25async def run_sync(26    func: Callable[[Unpack[PosArgsT]], T_Retval],27    *args: Unpack[PosArgsT],28    abandon_on_cancel: bool = False,29    cancellable: bool | None = None,30    limiter: CapacityLimiter | None = None,31) -> T_Retval:32    """33    Call the given function with the given arguments in a worker thread.34 35    If the ``abandon_on_cancel`` option is enabled and the task waiting for its36    completion is cancelled, the thread will still run its course but its37    return value (or any raised exception) will be ignored.38 39    :param func: a callable40    :param args: positional arguments for the callable41    :param abandon_on_cancel: ``True`` to abandon the thread (leaving it to run42        unchecked on own) if the host task is cancelled, ``False`` to ignore43        cancellations in the host task until the operation has completed in the worker44        thread45    :param cancellable: deprecated alias of ``abandon_on_cancel``; will override46        ``abandon_on_cancel`` if both parameters are passed47    :param limiter: capacity limiter to use to limit the total amount of threads running48        (if omitted, the default limiter is used)49    :raises NoEventLoopError: if no supported asynchronous event loop is running in the50        current thread51    :return: an awaitable that yields the return value of the function.52 53    """54    if cancellable is not None:55        abandon_on_cancel = cancellable56        warn(57            "The `cancellable=` keyword argument to `anyio.to_thread.run_sync` is "58            "deprecated since AnyIO 4.1.0; use `abandon_on_cancel=` instead",59            DeprecationWarning,60            stacklevel=2,61        )62 63    return await get_async_backend().run_sync_in_worker_thread(64        func, args, abandon_on_cancel=abandon_on_cancel, limiter=limiter65    )66 67 68def current_default_thread_limiter() -> CapacityLimiter:69    """70    Return the capacity limiter that is used by default to limit the number of71    concurrent threads.72 73    :return: a capacity limiter object74    :raises NoEventLoopError: if no supported asynchronous event loop is running in the75        current thread76 77    """78    return get_async_backend().current_default_thread_limiter()79