diff options
Diffstat (limited to 'venv/lib/python3.11/site-packages/uvloop/__init__.py')
-rw-r--r-- | venv/lib/python3.11/site-packages/uvloop/__init__.py | 168 |
1 files changed, 0 insertions, 168 deletions
diff --git a/venv/lib/python3.11/site-packages/uvloop/__init__.py b/venv/lib/python3.11/site-packages/uvloop/__init__.py deleted file mode 100644 index 9bb6592..0000000 --- a/venv/lib/python3.11/site-packages/uvloop/__init__.py +++ /dev/null @@ -1,168 +0,0 @@ -import asyncio as __asyncio -import typing as _typing -import sys as _sys -import warnings as _warnings - -from asyncio.events import BaseDefaultEventLoopPolicy as __BasePolicy - -from . import includes as __includes # NOQA -from .loop import Loop as __BaseLoop # NOQA -from ._version import __version__ # NOQA - - -__all__ = ('new_event_loop', 'install', 'EventLoopPolicy') - - -_T = _typing.TypeVar("_T") - - -class Loop(__BaseLoop, __asyncio.AbstractEventLoop): # type: ignore[misc] - pass - - -def new_event_loop() -> Loop: - """Return a new event loop.""" - return Loop() - - -def install() -> None: - """A helper function to install uvloop policy.""" - if _sys.version_info[:2] >= (3, 12): - _warnings.warn( - 'uvloop.install() is deprecated in favor of uvloop.run() ' - 'starting with Python 3.12.', - DeprecationWarning, - stacklevel=1, - ) - __asyncio.set_event_loop_policy(EventLoopPolicy()) - - -if _typing.TYPE_CHECKING: - def run( - main: _typing.Coroutine[_typing.Any, _typing.Any, _T], - *, - loop_factory: _typing.Optional[ - _typing.Callable[[], Loop] - ] = new_event_loop, - debug: _typing.Optional[bool]=None, - ) -> _T: - """The preferred way of running a coroutine with uvloop.""" -else: - def run(main, *, loop_factory=new_event_loop, debug=None, **run_kwargs): - """The preferred way of running a coroutine with uvloop.""" - - async def wrapper(): - # If `loop_factory` is provided we want it to return - # either uvloop.Loop or a subtype of it, assuming the user - # is using `uvloop.run()` intentionally. - loop = __asyncio._get_running_loop() - if not isinstance(loop, Loop): - raise TypeError('uvloop.run() uses a non-uvloop event loop') - return await main - - vi = _sys.version_info[:2] - - if vi <= (3, 10): - # Copied from python/cpython - - if __asyncio._get_running_loop() is not None: - raise RuntimeError( - "asyncio.run() cannot be called from a running event loop") - - if not __asyncio.iscoroutine(main): - raise ValueError( - "a coroutine was expected, got {!r}".format(main) - ) - - loop = loop_factory() - try: - __asyncio.set_event_loop(loop) - if debug is not None: - loop.set_debug(debug) - return loop.run_until_complete(wrapper()) - finally: - try: - _cancel_all_tasks(loop) - loop.run_until_complete(loop.shutdown_asyncgens()) - if hasattr(loop, 'shutdown_default_executor'): - loop.run_until_complete( - loop.shutdown_default_executor() - ) - finally: - __asyncio.set_event_loop(None) - loop.close() - - elif vi == (3, 11): - if __asyncio._get_running_loop() is not None: - raise RuntimeError( - "asyncio.run() cannot be called from a running event loop") - - with __asyncio.Runner( - loop_factory=loop_factory, - debug=debug, - **run_kwargs - ) as runner: - return runner.run(wrapper()) - - else: - assert vi >= (3, 12) - return __asyncio.run( - wrapper(), - loop_factory=loop_factory, - debug=debug, - **run_kwargs - ) - - -def _cancel_all_tasks(loop: __asyncio.AbstractEventLoop) -> None: - # Copied from python/cpython - - to_cancel = __asyncio.all_tasks(loop) - if not to_cancel: - return - - for task in to_cancel: - task.cancel() - - loop.run_until_complete( - __asyncio.gather(*to_cancel, return_exceptions=True) - ) - - for task in to_cancel: - if task.cancelled(): - continue - if task.exception() is not None: - loop.call_exception_handler({ - 'message': 'unhandled exception during asyncio.run() shutdown', - 'exception': task.exception(), - 'task': task, - }) - - -class EventLoopPolicy(__BasePolicy): - """Event loop policy. - - The preferred way to make your application use uvloop: - - >>> import asyncio - >>> import uvloop - >>> asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) - >>> asyncio.get_event_loop() - <uvloop.Loop running=False closed=False debug=False> - """ - - def _loop_factory(self) -> Loop: - return new_event_loop() - - if _typing.TYPE_CHECKING: - # EventLoopPolicy doesn't implement these, but since they are marked - # as abstract in typeshed, we have to put them in so mypy thinks - # the base methods are overridden. This is the same approach taken - # for the Windows event loop policy classes in typeshed. - def get_child_watcher(self) -> _typing.NoReturn: - ... - - def set_child_watcher( - self, watcher: _typing.Any - ) -> _typing.NoReturn: - ... |