Extra typing information for usage in IDE/ipython

Author: penguinolog

exec_helpers/_ssh_client_base.py

@@ -63,6 +63,7 @@
 _SSHConnChainT = typing.List[typing.Tuple[SSHConfig, ssh_auth.SSHAuth]]
 _OptionalTimeoutT = typing.Union[int, float, None]
 _OptionalStdinT = typing.Union[bytes, str, bytearray, None]
+_ExitCodeT = typing.Union[int, proc_enums.ExitCodes]
 
 
 class RetryOnExceptions(tenacity.retry_if_exception):  # type: ignore
@@ -78,7 +79,7 @@ def __init__(
         :param retry_on: Exceptions to retry on
         :param reraise: Exceptions, which should be reraised, even if subclasses retry_on
         """
-        super(RetryOnExceptions, self).__init__(lambda e: isinstance(e, retry_on) and not isinstance(e, reraise))
+        super().__init__(lambda e: isinstance(e, retry_on) and not isinstance(e, reraise))
 
 
 # noinspection PyTypeHints
@@ -88,25 +89,25 @@ class SshExecuteAsyncResult(api.ExecuteAsyncResult):
     @property
     def interface(self) -> paramiko.Channel:
         """Override original NamedTuple with proper typing."""
-        return super(SshExecuteAsyncResult, self).interface
+        return super().interface
 
     @property
     def stdin(self) -> paramiko.ChannelFile:  # type: ignore
         """Override original NamedTuple with proper typing."""
-        return super(SshExecuteAsyncResult, self).stdin
+        return super().stdin
 
     @property
     def stderr(self) -> typing.Optional[paramiko.ChannelFile]:  # type: ignore
         """Override original NamedTuple with proper typing."""
-        return super(SshExecuteAsyncResult, self).stderr
+        return super().stderr
 
     @property
     def stdout(self) -> typing.Optional[paramiko.ChannelFile]:  # type: ignore
         """Override original NamedTuple with proper typing."""
-        return super(SshExecuteAsyncResult, self).stdout
+        return super().stdout
 
 
-class _SudoContext:
+class _SudoContext(typing.ContextManager[None]):
     """Context manager for call commands with sudo."""
 
     __slots__ = ("__ssh", "__sudo_status", "__enforce")
@@ -132,7 +133,7 @@ def __exit__(self, exc_type: typing.Any, exc_val: typing.Any, exc_tb: typing.Any
         self.__ssh.sudo_mode = self.__sudo_status
 
 
-class _KeepAliveContext:
+class _KeepAliveContext(typing.ContextManager[None]):
     """Context manager for keepalive management."""
 
     __slots__ = ("__ssh", "__keepalive_status", "__enforce")
@@ -146,12 +147,12 @@ def __init__(self, ssh: "SSHClientBase", enforce: int) -> None:
         :type enforce: int
         """
         self.__ssh: "SSHClientBase" = ssh
-        self.__keepalive_status: int = ssh.keepalive_mode
+        self.__keepalive_status: int = ssh.keepalive_period
         self.__enforce: int = enforce
 
     def __enter__(self) -> None:
         self.__ssh.__enter__()
-        self.__keepalive_status = self.__ssh.keepalive_mode
+        self.__keepalive_status = self.__ssh.keepalive_period
         self.__ssh.keepalive_mode = self.__enforce
 
     def __exit__(self, exc_type: typing.Any, exc_val: typing.Any, exc_tb: typing.Any) -> None:
@@ -524,7 +525,7 @@ def __exit__(self, exc_type: typing.Any, exc_val: typing.Any, exc_tb: typing.Any
         """
         if not self.__keepalive_period:
             self.close()
-        super(SSHClientBase, self).__exit__(exc_type, exc_val, exc_tb)
+        super().__exit__(exc_type, exc_val, exc_tb)
 
     @property
     def sudo_mode(self) -> bool:
@@ -930,7 +931,7 @@ def check_call(  # pylint: disable=arguments-differ
         verbose: bool = False,
         timeout: _OptionalTimeoutT = constants.DEFAULT_TIMEOUT,
         error_info: typing.Optional[str] = None,
-        expected: typing.Iterable[typing.Union[int, proc_enums.ExitCodes]] = (proc_enums.EXPECTED,),
+        expected: typing.Iterable[_ExitCodeT] = (proc_enums.EXPECTED,),
         raise_on_err: bool = True,
         *,
         log_mask_re: typing.Optional[str] = None,
@@ -1011,7 +1012,7 @@ def check_stderr(  # pylint: disable=arguments-differ
         error_info: typing.Optional[str] = None,
         raise_on_err: bool = True,
         *,
-        expected: typing.Iterable[typing.Union[int, proc_enums.ExitCodes]] = (proc_enums.EXPECTED,),
+        expected: typing.Iterable[_ExitCodeT] = (proc_enums.EXPECTED,),
         log_mask_re: typing.Optional[str] = None,
         stdin: _OptionalStdinT = None,
         open_stdout: bool = True,
@@ -1262,7 +1263,7 @@ def execute_together(
         remotes: typing.Iterable["SSHClientBase"],
         command: str,
         timeout: _OptionalTimeoutT = constants.DEFAULT_TIMEOUT,
-        expected: typing.Iterable[typing.Union[int, proc_enums.ExitCodes]] = (proc_enums.EXPECTED,),
+        expected: typing.Iterable[_ExitCodeT] = (proc_enums.EXPECTED,),
         raise_on_err: bool = True,
         *,
         stdin: _OptionalStdinT = None,
@@ -1349,9 +1350,7 @@ def get_result(remote: "SSHClientBase") -> exec_result.ExecResult:
             async_result.interface.close()
             return res
 
-        prep_expected: typing.Tuple[typing.Union[int, proc_enums.ExitCodes], ...] = proc_enums.exit_codes_to_enums(
-            expected
-        )
+        prep_expected: typing.Sequence[_ExitCodeT] = proc_enums.exit_codes_to_enums(expected)
         log_level: int = logging.INFO if verbose else logging.DEBUG
 
         futures: typing.Dict["SSHClientBase", "concurrent.futures.Future[exec_result.ExecResult]"] = {

exec_helpers/api.py

@@ -49,10 +49,11 @@
 )
 _OptionalTimeoutT = typing.Union[int, float, None]
 _OptionalStdinT = typing.Union[bytes, str, bytearray, None]
+_ExitCodeT = typing.Union[int, proc_enums.ExitCodes]
 
 
 # noinspection PyProtectedMember
-class _ChRootContext:
+class _ChRootContext(typing.ContextManager[None]):
     """Context manager for call commands with chroot.
 
     .. versionadded:: 4.1.0
@@ -444,7 +445,7 @@ def check_call(
         verbose: bool = False,
         timeout: _OptionalTimeoutT = constants.DEFAULT_TIMEOUT,
         error_info: typing.Optional[str] = None,
-        expected: typing.Iterable[typing.Union[int, proc_enums.ExitCodes]] = (proc_enums.EXPECTED,),
+        expected: typing.Iterable[_ExitCodeT] = (proc_enums.EXPECTED,),
         raise_on_err: bool = True,
         *,
         log_mask_re: typing.Optional[str] = None,
@@ -490,9 +491,7 @@ def check_call(
         .. versionchanged:: 3.2.0 Exception class can be substituted
         .. versionchanged:: 3.4.0 Expected is not optional, defaults os dependent
         """
-        expected_codes: typing.Tuple[typing.Union[int, proc_enums.ExitCodes], ...] = proc_enums.exit_codes_to_enums(
-            expected
-        )
+        expected_codes: typing.Sequence[_ExitCodeT] = proc_enums.exit_codes_to_enums(expected)
         result: exec_result.ExecResult = self.execute(
             command,
             verbose=verbose,
@@ -522,7 +521,7 @@ def check_stderr(
         error_info: typing.Optional[str] = None,
         raise_on_err: bool = True,
         *,
-        expected: typing.Iterable[typing.Union[int, proc_enums.ExitCodes]] = (proc_enums.EXPECTED,),
+        expected: typing.Iterable[_ExitCodeT] = (proc_enums.EXPECTED,),
         log_mask_re: typing.Optional[str] = None,
         stdin: _OptionalStdinT = None,
         open_stdout: bool = True,
@@ -593,7 +592,7 @@ def _handle_stderr(
         result: exec_result.ExecResult,
         error_info: typing.Optional[str],
         raise_on_err: bool,
-        expected: typing.Iterable[typing.Union[int, proc_enums.ExitCodes]],
+        expected: typing.Iterable[_ExitCodeT],
         exception_class: "typing.Type[exceptions.CalledProcessError]",
     ) -> exec_result.ExecResult:
         """Internal check_stderr logic (synchronous)."""

exec_helpers/async_api/api.py

@@ -34,8 +34,13 @@
 from exec_helpers import proc_enums
 
 
+_OptionalTimeoutT = typing.Union[int, float, None]
+_OptionalStdinT = typing.Union[bytes, str, bytearray, None]
+_ExitCodeT = typing.Union[int, proc_enums.ExitCodes]
+
+
 # noinspection PyProtectedMember
-class _ChRootContext(api._ChRootContext):  # pylint: disable=protected-access
+class _ChRootContext(api._ChRootContext, typing.AsyncContextManager[None]):  # pylint: disable=protected-access
     """Async extension for chroot."""
 
     def __init__(self, conn: "ExecHelper", path: typing.Optional[typing.Union[str, pathlib.Path]] = None) -> None:
@@ -46,7 +51,7 @@ def __init__(self, conn: "ExecHelper", path: typing.Optional[typing.Union[str, p
         :param path: chroot path or None for no chroot
         :type path: typing.Optional[typing.Union[str, pathlib.Path]]
         """
-        super(_ChRootContext, self).__init__(conn=conn, path=path)
+        super().__init__(conn=conn, path=path)
 
     async def __aenter__(self) -> None:
         await self._conn.__aenter__()
@@ -72,7 +77,7 @@ def __init__(self, log_mask_re: typing.Optional[str] = None, *, logger: logging.
                             all MATCHED groups will be replaced by '<*masked*>'
         :type log_mask_re: typing.Optional[str]
         """
-        super(ExecHelper, self).__init__(logger=logger, log_mask_re=log_mask_re)
+        super().__init__(logger=logger, log_mask_re=log_mask_re)
         self.__alock: typing.Optional[asyncio.Lock] = None
 
     def __enter__(self) -> "ExecHelper":  # pylint: disable=useless-super-delegation
@@ -108,11 +113,11 @@ async def _exec_command(  # type: ignore
         self,
         command: str,
         async_result: api.ExecuteAsyncResult,
-        timeout: typing.Union[int, float, None],
+        timeout: _OptionalTimeoutT,
         *,
         verbose: bool = False,
         log_mask_re: typing.Optional[str] = None,
-        stdin: typing.Union[bytes, str, bytearray, None] = None,
+        stdin: _OptionalStdinT = None,
         **kwargs: typing.Any,
     ) -> exec_result.ExecResult:
         """Get exit status from channel with timeout.
@@ -181,10 +186,10 @@ async def execute(  # type: ignore
         self,
         command: str,
         verbose: bool = False,
-        timeout: typing.Union[int, float, None] = constants.DEFAULT_TIMEOUT,
+        timeout: _OptionalTimeoutT = constants.DEFAULT_TIMEOUT,
         *,
         log_mask_re: typing.Optional[str] = None,
-        stdin: typing.Union[bytes, str, bytearray, None] = None,
+        stdin: _OptionalStdinT = None,
         open_stdout: bool = True,
         open_stderr: bool = True,
         **kwargs: typing.Any,
@@ -247,10 +252,10 @@ async def __call__(  # type: ignore
         self,
         command: str,
         verbose: bool = False,
-        timeout: typing.Union[int, float, None] = constants.DEFAULT_TIMEOUT,
+        timeout: _OptionalTimeoutT = constants.DEFAULT_TIMEOUT,
         *,
         log_mask_re: typing.Optional[str] = None,
-        stdin: typing.Union[bytes, str, bytearray, None] = None,
+        stdin: _OptionalStdinT = None,
         open_stdout: bool = True,
         open_stderr: bool = True,
         **kwargs: typing.Any,
@@ -295,13 +300,13 @@ async def check_call(  # type: ignore
         self,
         command: str,
         verbose: bool = False,
-        timeout: typing.Union[int, float, None] = constants.DEFAULT_TIMEOUT,
+        timeout: _OptionalTimeoutT = constants.DEFAULT_TIMEOUT,
         error_info: typing.Optional[str] = None,
-        expected: typing.Iterable[typing.Union[int, proc_enums.ExitCodes]] = (proc_enums.EXPECTED,),
+        expected: typing.Iterable[_ExitCodeT] = (proc_enums.EXPECTED,),
         raise_on_err: bool = True,
         *,
         log_mask_re: typing.Optional[str] = None,
-        stdin: typing.Union[bytes, str, bytearray, None] = None,
+        stdin: _OptionalStdinT = None,
         open_stdout: bool = True,
         open_stderr: bool = True,
         exception_class: "typing.Type[exceptions.CalledProcessError]" = exceptions.CalledProcessError,
@@ -341,9 +346,7 @@ async def check_call(  # type: ignore
 
         .. versionchanged:: 3.4.0 Expected is not optional, defaults os dependent
         """
-        expected_codes: typing.Tuple[typing.Union[int, proc_enums.ExitCodes], ...] = proc_enums.exit_codes_to_enums(
-            expected
-        )
+        expected_codes: typing.Sequence[_ExitCodeT] = proc_enums.exit_codes_to_enums(expected)
         result: exec_result.ExecResult = await self.execute(
             command,
             verbose=verbose,
@@ -369,13 +372,13 @@ async def check_stderr(  # type: ignore
         self,
         command: str,
         verbose: bool = False,
-        timeout: typing.Union[int, float, None] = constants.DEFAULT_TIMEOUT,
+        timeout: _OptionalTimeoutT = constants.DEFAULT_TIMEOUT,
         error_info: typing.Optional[str] = None,
         raise_on_err: bool = True,
         *,
-        expected: typing.Iterable[typing.Union[int, proc_enums.ExitCodes]] = (proc_enums.EXPECTED,),
+        expected: typing.Iterable[_ExitCodeT] = (proc_enums.EXPECTED,),
         log_mask_re: typing.Optional[str] = None,
-        stdin: typing.Union[bytes, str, bytearray, None] = None,
+        stdin: _OptionalStdinT = None,
         open_stdout: bool = True,
         open_stderr: bool = True,
         exception_class: "typing.Type[exceptions.CalledProcessError]" = exceptions.CalledProcessError,