diff --git a/lib/ansible/executor/task_queue_manager.py b/lib/ansible/executor/task_queue_manager.py index c48f3b9bb00..84450b60e69 100644 --- a/lib/ansible/executor/task_queue_manager.py +++ b/lib/ansible/executor/task_queue_manager.py @@ -58,8 +58,6 @@ STDERR_FILENO = 2 display = Display() -_T = t.TypeVar('_T') - @dataclasses.dataclass(frozen=True, kw_only=True, slots=True) class CallbackSend: @@ -413,7 +411,7 @@ class TaskQueueManager: return defunct @staticmethod - def _first_arg_of_type(value_type: t.Type[_T], args: t.Sequence) -> _T | None: + def _first_arg_of_type[T](value_type: t.Type[T], args: t.Sequence) -> T | None: return next((arg for arg in args if isinstance(arg, value_type)), None) @lock_decorator(attr='_callback_lock') diff --git a/lib/ansible/plugins/callback/__init__.py b/lib/ansible/plugins/callback/__init__.py index aa8beaea290..ea675ee444f 100644 --- a/lib/ansible/plugins/callback/__init__.py +++ b/lib/ansible/plugins/callback/__init__.py @@ -60,9 +60,6 @@ _YAML_BREAK_CHARS = '\n\x85\u2028\u2029' # NL, NEL, LS, PS _SPACE_BREAK_RE = re.compile(fr' +([{_YAML_BREAK_CHARS}])') -_T_callable = t.TypeVar("_T_callable", bound=t.Callable) - - class _AnsibleCallbackDumper(_dumper.AnsibleDumper): def __init__(self, *args, lossy: bool = False, **kwargs): super().__init__(*args, **kwargs) diff --git a/lib/ansible/plugins/connection/__init__.py b/lib/ansible/plugins/connection/__init__.py index 553235884fd..0078cd9c52e 100644 --- a/lib/ansible/plugins/connection/__init__.py +++ b/lib/ansible/plugins/connection/__init__.py @@ -6,7 +6,6 @@ from __future__ import annotations import collections.abc as c import fcntl -import io import os import shlex import typing as t @@ -16,7 +15,7 @@ from functools import wraps from ansible import constants as C from ansible.errors import AnsibleValueOmittedError -from ansible.module_utils.common.text.converters import to_bytes, to_text +from ansible.module_utils.common.text.converters import to_text from ansible.playbook.play_context import PlayContext from ansible.plugins import AnsiblePlugin from ansible.plugins.become import BecomeBase @@ -32,9 +31,6 @@ __all__ = ['ConnectionBase', 'ensure_connect'] BUFSIZE = 65536 -P = t.ParamSpec('P') -T = t.TypeVar('T') - class ConnectionKwargs(t.TypedDict): task_uuid: str @@ -42,7 +38,7 @@ class ConnectionKwargs(t.TypedDict): shell: t.NotRequired[ShellBase] -def ensure_connect( +def ensure_connect[T, **P]( func: c.Callable[t.Concatenate[ConnectionBase, P], T], ) -> c.Callable[t.Concatenate[ConnectionBase, P], T]: @wraps(func) @@ -135,7 +131,7 @@ class ConnectionBase(AnsiblePlugin): pass @abstractmethod - def _connect(self: T) -> T: + def _connect[T](self: T) -> T: """Connect to the host we've been initialized with""" @ensure_connect diff --git a/lib/ansible/plugins/connection/ssh.py b/lib/ansible/plugins/connection/ssh.py index a4f7fedcd1c..b9b6c1356a1 100644 --- a/lib/ansible/plugins/connection/ssh.py +++ b/lib/ansible/plugins/connection/ssh.py @@ -459,8 +459,6 @@ else: display = Display() -P = t.ParamSpec('P') - # error messages that indicate 255 return code is not from ssh itself. b_NOT_SSH_ERRORS = (b'Traceback (most recent call last):', # Python-2.6 when there's an exception # while invoking a script via -m @@ -547,7 +545,7 @@ def _handle_error( display.vvv(msg, host=host) -def _ssh_retry( +def _ssh_retry[**P]( func: c.Callable[t.Concatenate[Connection, P], tuple[int, bytes, bytes]], ) -> c.Callable[t.Concatenate[Connection, P], tuple[int, bytes, bytes]]: """ diff --git a/lib/ansible/template/__init__.py b/lib/ansible/template/__init__.py index 7e3a93bf3ba..c2ecb92313b 100644 --- a/lib/ansible/template/__init__.py +++ b/lib/ansible/template/__init__.py @@ -25,7 +25,6 @@ if _t.TYPE_CHECKING: # pragma: nocover _display: _t.Final[_Display] = _Display() _UNSET = _t.cast(_t.Any, object()) -_TTrustable = _t.TypeVar('_TTrustable', bound=str | _io.IOBase | _t.TextIO | _t.BinaryIO) _TRUSTABLE_TYPES = (str, _io.IOBase) AnsibleUndefined = _jinja_common.UndefinedMarker @@ -361,7 +360,7 @@ def generate_ansible_template_vars( return _template_vars.generate_ansible_template_vars(path=path, fullpath=fullpath, dest_path=dest_path, include_ansible_managed=True) -def trust_as_template(value: _TTrustable) -> _TTrustable: +def trust_as_template[T: str | _io.IOBase | _t.TextIO | _t.BinaryIO](value: T) -> T: """ Returns `value` tagged as trusted for templating. Raises a `TypeError` if `value` is not a supported type. @@ -385,10 +384,7 @@ def is_trusted_as_template(value: object) -> bool: return isinstance(value, _TRUSTABLE_TYPES) and _tags.TrustedAsTemplate.is_tagged_on(value) -_TCallable = _t.TypeVar('_TCallable', bound=_t.Callable) - - -def accept_args_markers(plugin: _TCallable) -> _TCallable: +def accept_args_markers[T: _t.Callable](plugin: T) -> T: """ A decorator to mark a Jinja plugin as capable of handling `Marker` values for its top-level arguments. Non-decorated plugin invocation is skipped when a top-level argument is a `Marker`, with the first such value substituted as the plugin result. @@ -399,7 +395,7 @@ def accept_args_markers(plugin: _TCallable) -> _TCallable: return plugin -def accept_lazy_markers(plugin: _TCallable) -> _TCallable: +def accept_lazy_markers[T: _t.Callable](plugin: T) -> T: """ A decorator to mark a Jinja plugin as capable of handling `Marker` values retrieved from lazy containers. Non-decorated plugins will trigger a `MarkerError` exception when attempting to retrieve a `Marker` from a lazy container. diff --git a/lib/ansible/utils/display.py b/lib/ansible/utils/display.py index 8650dcb91af..a0abd5dccd5 100644 --- a/lib/ansible/utils/display.py +++ b/lib/ansible/utils/display.py @@ -67,8 +67,6 @@ if t.TYPE_CHECKING: # avoid circular import at runtime from ansible.executor.task_queue_manager import FinalQueue -P = t.ParamSpec('P') - _LIBC = ctypes.cdll.LoadLibrary(ctypes.util.find_library('c')) # Set argtypes, to avoid segfault if the wrong type is provided, # restype is assumed to be c_int @@ -388,7 +386,7 @@ class Display(metaclass=Singleton): self.b_cowsay = b_cow_path @staticmethod - def _proxy( + def _proxy[**P]( func: c.Callable[t.Concatenate[Display, P], None] ) -> c.Callable[..., None]: @wraps(func) diff --git a/lib/ansible/utils/unsafe_proxy.py b/lib/ansible/utils/unsafe_proxy.py index 9ecd6aa5999..1a2c6d04b24 100644 --- a/lib/ansible/utils/unsafe_proxy.py +++ b/lib/ansible/utils/unsafe_proxy.py @@ -12,12 +12,8 @@ from ansible.module_utils.common.collections import is_sequence from ansible._internal._datatag._tags import TrustedAsTemplate from ansible.module_utils.six import binary_type, text_type -import typing as t - __all__ = ['AnsibleUnsafe', 'wrap_var'] -T = t.TypeVar('T') - class AnsibleUnsafe: def __new__(cls, value): diff --git a/packaging/release.py b/packaging/release.py index 59077ec5eb9..e4c89bd84e1 100755 --- a/packaging/release.py +++ b/packaging/release.py @@ -42,9 +42,6 @@ from packaging.version import Version, InvalidVersion # region CLI Framework -C = t.TypeVar("C", bound=t.Callable[..., None]) - - def path_to_str(value: t.Any) -> str: """Return the given value converted to a string suitable for use as a command line argument.""" return f"{value}/" if isinstance(value, pathlib.Path) and value.is_dir() else str(value) @@ -188,7 +185,7 @@ class CommandFramework: self.arguments = kwargs self.parsed_arguments: argparse.Namespace | None = None - def __call__(self, func: C) -> C: + def __call__[T: t.Callable[..., None]](self, func: T) -> T: """Register the decorated function as a CLI command.""" self.commands.append(func) return func