CPython 3.14-3.16 compat

* Add deprecation in docstring
* Stop using private API, refs python/cpython#131148
* Use AbstractEventLoop instead, after @Vizonex's suggestion

Author: fantix

pyproject.toml

@@ -2,7 +2,7 @@
 name = "uvloop"
 description = "Fast implementation of asyncio event loop on top of libuv"
 authors = [{name = "Yury Selivanov", email = "[email protected]"}]
-requires-python = '>=3.8.0'
+requires-python = '>=3.8.1'
 readme = "README.rst"
 license = {text = "MIT License"}
 dynamic = ["version"]
@@ -38,9 +38,9 @@ test = [
     # their combination breaks too often
     # (example breakage: https://gitlab.com/pycqa/flake8/issues/427)
     'aiohttp>=3.10.5',
-    'flake8~=5.0',
+    'flake8~=6.1',
     'psutil',
-    'pycodestyle~=2.9.0',
+    'pycodestyle~=2.11.0',
     'pyOpenSSL~=23.0.0',
     'mypy>=0.800',
 ]

uvloop/__init__.py

@@ -3,18 +3,12 @@
 import sys as _sys
 import warnings as _warnings
 
-try:
-    from asyncio.events import BaseDefaultEventLoopPolicy as __BasePolicy
-except ImportError:
-    # https://github.com/python/cpython/issues/131148
-    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')
+__all__: tuple[str, ...] = ('new_event_loop',)
 
 
 _T = _typing.TypeVar("_T")
@@ -29,18 +23,6 @@ def new_event_loop() -> 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],
@@ -143,30 +125,91 @@ def _cancel_all_tasks(loop: __asyncio.AbstractEventLoop) -> None:
             })
 
 
-class EventLoopPolicy(__BasePolicy):
-    """Event loop policy.
+if _sys.version_info[:2] < (3, 16):
+    import threading as _threading
 
-    The preferred way to make your application use uvloop:
+    __all__ += ('install', 'EventLoopPolicy')
 
-    >>> import asyncio
-    >>> import uvloop
-    >>> asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
-    >>> asyncio.get_event_loop()
-    <uvloop.Loop running=False closed=False debug=False>
-    """
+    def install() -> None:
+        """A helper function to install uvloop policy.
 
-    def _loop_factory(self) -> Loop:
-        return new_event_loop()
+        This function is deprecated and will be removed in Python 3.16.
+        Use `uvloop.run()` instead.
+        """
+        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())
+
+    class EventLoopPolicy(
+        __asyncio.AbstractEventLoopPolicy  # type: ignore[name-defined,misc]
+    ):
+        """Event loop policy for uvloop.
+
+        This class is deprecated and will be removed in Python 3.16.
+        Use `uvloop.run()` instead.
+
+        >>> 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:
+                ...
+
+        class _Local(_threading.local):
+            _loop: _typing.Optional[Loop] = None
+
+        def __init__(self) -> None:
+            self._local = self._Local()
+
+        def get_event_loop(self) -> Loop:
+            """Get the event loop for the current context.
+
+            Returns an instance of EventLoop or raises an exception.
+            """
+            if self._local._loop is None:
+                raise RuntimeError(
+                    'There is no current event loop in thread %r.'
+                    % _threading.current_thread().name
+                )
+
+            return self._local._loop
+
+        def set_event_loop(self, loop: Loop) -> None:
+            """Set the event loop."""
+            if loop is not None and not isinstance(
+                loop, __asyncio.AbstractEventLoop
+            ):
+                raise TypeError(
+                    f"loop must be an instance of AbstractEventLoop or None, "
+                    f"not '{type(loop).__name__}'"
+                )
+            self._local._loop = 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 new_event_loop(self) -> Loop:
+            """Create a new event loop.
 
-        def set_child_watcher(
-            self, watcher: _typing.Any
-        ) -> _typing.NoReturn:
-            ...
+            You must call set_event_loop() to make this the current event
+            loop.
+            """
+            return self._loop_factory()