codekingpro/portable-devtools

1from __future__ import annotations2 3import asyncio4import sys5from collections.abc import Callable, Coroutine6from typing import Any, TypeVar7 8__all__ = ["asyncio_run", "iscoroutinefunction"]9 10if sys.version_info >= (3, 14):11    from inspect import iscoroutinefunction12else:13    from asyncio import iscoroutinefunction14 15_T = TypeVar("_T")16 17if sys.version_info >= (3, 12):18    asyncio_run = asyncio.run19elif sys.version_info >= (3, 11):20 21    def asyncio_run(22        main: Coroutine[Any, Any, _T],23        *,24        debug: bool = False,25        loop_factory: Callable[[], asyncio.AbstractEventLoop] | None = None,26    ) -> _T:27        # asyncio.run from Python 3.1228        # https://docs.python.org/3/license.html#psf-license29        with asyncio.Runner(debug=debug, loop_factory=loop_factory) as runner:30            return runner.run(main)31 32else:33    # modified version of asyncio.run from Python 3.10 to add loop_factory kwarg34    # https://docs.python.org/3/license.html#psf-license35    def asyncio_run(36        main: Coroutine[Any, Any, _T],37        *,38        debug: bool = False,39        loop_factory: Callable[[], asyncio.AbstractEventLoop] | None = None,40    ) -> _T:41        try:42            asyncio.get_running_loop()43        except RuntimeError:44            pass45        else:46            raise RuntimeError("asyncio.run() cannot be called from a running event loop")47 48        if not asyncio.iscoroutine(main):49            raise ValueError(f"a coroutine was expected, got {main!r}")50 51        if loop_factory is None:52            loop = asyncio.new_event_loop()53        else:54            loop = loop_factory()55        try:56            if loop_factory is None:57                asyncio.set_event_loop(loop)58            if debug is not None:59                loop.set_debug(debug)60            return loop.run_until_complete(main)61        finally:62            try:63                _cancel_all_tasks(loop)64                loop.run_until_complete(loop.shutdown_asyncgens())65                loop.run_until_complete(loop.shutdown_default_executor())66            finally:67                if loop_factory is None:68                    asyncio.set_event_loop(None)69                loop.close()70 71    def _cancel_all_tasks(loop: asyncio.AbstractEventLoop) -> None:72        to_cancel = asyncio.all_tasks(loop)73        if not to_cancel:74            return75 76        for task in to_cancel:77            task.cancel()78 79        loop.run_until_complete(asyncio.gather(*to_cancel, return_exceptions=True))80 81        for task in to_cancel:82            if task.cancelled():83                continue84            if task.exception() is not None:85                loop.call_exception_handler(86                    {87                        "message": "unhandled exception during asyncio.run() shutdown",88                        "exception": task.exception(),89                        "task": task,90                    }91                )92