From: Aleš Mrázek Date: Sun, 21 Dec 2025 21:11:59 +0000 (+0100) Subject: utils/compat: python3.8 cleanup X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=48b743d440a3786b221eda31121043ebca676dd5;p=thirdparty%2Fknot-resolver.git utils/compat: python3.8 cleanup - removed old unused/unnecessary code - removed unused typing.py module --- diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 000000000..9b388533a --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,7 @@ +{ + "python.testing.pytestArgs": [ + "tests" + ], + "python.testing.unittestEnabled": false, + "python.testing.pytestEnabled": true +} \ No newline at end of file diff --git a/python/knot_resolver/utils/compat/__init__.py b/python/knot_resolver/utils/compat/__init__.py index 52ffaa9cb..e69de29bb 100644 --- a/python/knot_resolver/utils/compat/__init__.py +++ b/python/knot_resolver/utils/compat/__init__.py @@ -1,3 +0,0 @@ -from . import asyncio, typing - -__all__ = ["asyncio", "typing"] diff --git a/python/knot_resolver/utils/compat/asyncio.py b/python/knot_resolver/utils/compat/asyncio.py index 95225c011..62372227c 100644 --- a/python/knot_resolver/utils/compat/asyncio.py +++ b/python/knot_resolver/utils/compat/asyncio.py @@ -1,13 +1,8 @@ -# We disable pylint checks, because it can't find methods in newer Python versions. -# -# pylint: disable=no-member - import asyncio import functools import logging import sys -from asyncio import AbstractEventLoop, coroutines, events, tasks -from typing import Any, Awaitable, Callable, Coroutine, Optional, TypeVar +from typing import Any, Callable, Coroutine, TypeVar logger = logging.getLogger(__name__) @@ -17,8 +12,7 @@ T = TypeVar("T") async def to_thread(func: Callable[..., T], *args: Any, **kwargs: Any) -> T: # version 3.9 and higher, call directly if sys.version_info >= (3, 9): - return await asyncio.to_thread(func, *args, **kwargs) # type: ignore[attr-defined] - + return await asyncio.to_thread(func, *args, **kwargs) # earlier versions, run with default executor loop = asyncio.get_event_loop() pfunc = functools.partial(func, *args, **kwargs) @@ -32,94 +26,16 @@ def async_in_a_thread(func: Callable[..., T]) -> Callable[..., Coroutine[None, N return wrapper -def create_task(coro: Awaitable[T], name: Optional[str] = None) -> "asyncio.Task[T]": - # version 3.8 and higher, call directly - if sys.version_info >= (3, 8): - # pylint: disable=unexpected-keyword-arg - return asyncio.create_task(coro, name=name) # type: ignore[attr-defined,arg-type,call-arg] - - # version 3.7 and higher, call directly without the name argument - if sys.version_info >= (3, 8): - return asyncio.create_task(coro) # type: ignore[attr-defined,arg-type] - - # earlier versions, use older function - return asyncio.ensure_future(coro) - - -def is_event_loop_running() -> bool: - loop = events._get_running_loop() # noqa: SLF001 - return loop is not None and loop.is_running() - - -def run(coro: Awaitable[T], debug: Optional[bool] = None) -> T: - # Adapted version of this: - # https://github.com/python/cpython/blob/3.9/Lib/asyncio/runners.py#L8 - - # version 3.7 and higher, call directly - # disabled due to incompatibilities - if sys.version_info >= (3, 7): - return asyncio.run(coro, debug=debug) # type: ignore[attr-defined,arg-type] - - # earlier versions, use backported version of the function - if events._get_running_loop() is not None: # noqa: SLF001 - raise RuntimeError("asyncio.run() cannot be called from a running event loop") - - if not coroutines.iscoroutine(coro): - raise ValueError(f"a coroutine was expected, got {repr(coro)}") - - loop = events.new_event_loop() - try: - events.set_event_loop(loop) - if debug is not None: - loop.set_debug(debug) - return loop.run_until_complete(coro) - 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()) # type: ignore[attr-defined] - finally: - events.set_event_loop(None) - loop.close() - - -def _cancel_all_tasks(loop: AbstractEventLoop) -> None: - # Backported from: - # https://github.com/python/cpython/blob/3.9/Lib/asyncio/runners.py#L55-L74 - # - to_cancel = tasks.all_tasks(loop) - if not to_cancel: - return - - for task in to_cancel: - task.cancel() - - if sys.version_info >= (3, 7): - # since 3.7, the loop argument is implicitely the running loop - # since 3.10, the loop argument is removed - loop.run_until_complete(tasks.gather(*to_cancel, return_exceptions=True)) - else: - loop.run_until_complete(tasks.gather(*to_cancel, loop=loop, return_exceptions=True)) # type: ignore[call-overload] - - 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, - } - ) - - def add_async_signal_handler(signal: int, callback: Callable[[], Coroutine[Any, Any, None]]) -> None: loop = asyncio.get_event_loop() - loop.add_signal_handler(signal, lambda: create_task(callback())) + loop.add_signal_handler(signal, lambda: asyncio.create_task(callback())) def remove_signal_handler(signal: int) -> bool: loop = asyncio.get_event_loop() return loop.remove_signal_handler(signal) + + +def is_event_loop_running() -> bool: + loop = asyncio.events._get_running_loop() # noqa: SLF001 + return loop is not None and loop.is_running() diff --git a/python/knot_resolver/utils/compat/typing.py b/python/knot_resolver/utils/compat/typing.py deleted file mode 100644 index 15654d345..000000000 --- a/python/knot_resolver/utils/compat/typing.py +++ /dev/null @@ -1,8 +0,0 @@ -# The 'typing.Pattern' is deprecated since python 3.8 and is removed in version 3.12. -# https://docs.python.org/3.9/library/typing.html#typing.Pattern -try: - from typing import Pattern -except ImportError: - from re import Pattern - -__all__ = ["Pattern"]