Eliminate more TypeVar and ParamSpec usage (#85602)

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')

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)

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

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]]:
     """

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.

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)

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):

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