better typing information

python-useful-helpers · Nov 27, 2019 · 6ca31c7 · 6ca31c7
1 parent 1896e81
commit 6ca31c7
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 16 deletions.
diff --git a/doc/source/Subprocess.rst b/doc/source/Subprocess.rst
@@ -173,19 +173,19 @@ API: Subprocess
 
     .. py:attribute:: interface
 
-        ``subprocess.Popen``
+        ``subprocess.Popen[bytes]``
 
     .. py:attribute:: stdin
 
-        ``typing.Optional[typing.IO]``
+        ``typing.Optional[typing.IO[bytes]]``
 
     .. py:attribute:: stderr
 
-        ``typing.Optional[typing.IO]``
+        ``typing.Optional[typing.IO[bytes]]``
 
     .. py:attribute:: stdout
 
-        ``typing.Optional[typing.IO]``
+        ``typing.Optional[typing.IO[bytes]]``
 
     .. py:attribute:: started
 

diff --git a/exec_helpers/ssh_auth.py b/exec_helpers/ssh_auth.py
@@ -179,6 +179,7 @@ def connect(
         for index, key in sorted(enumerate(self.__keys), key=lambda i_k: i_k[0] != self.__key_index):
             kwargs["pkey"] = key
             try:
+                # noinspection PyTypeChecker
                 client.connect(
                     hostname=hostname,
                     port=port,
@@ -240,6 +241,7 @@ def __deepcopy__(self, memo: typing.Any) -> "SSHAuth":
         :param memo: copy.deeepcopy() memodict
         :return: re-constructed copy of current class
         """
+        # noinspection PyTypeChecker
         return self.__class__(
             username=self.username,
             password=self.__password,
@@ -249,6 +251,7 @@ def __deepcopy__(self, memo: typing.Any) -> "SSHAuth":
 
     def __copy__(self) -> "SSHAuth":
         """Copy self."""
+        # noinspection PyTypeChecker
         return self.__class__(
             username=self.username, password=self.__password, key=self.__keys[self.__key_index], keys=self.__keys
         )

diff --git a/exec_helpers/subprocess_runner.py b/exec_helpers/subprocess_runner.py
@@ -42,22 +42,22 @@ class SubprocessExecuteAsyncResult(api.ExecuteAsyncResult):
     """Override original NamedTuple with proper typing."""
 
     @property
-    def interface(self) -> "subprocess.Popen[str]":
+    def interface(self) -> "subprocess.Popen[bytes]":
         """Override original NamedTuple with proper typing."""
         return super(SubprocessExecuteAsyncResult, self).interface  # type: ignore
 
     @property
-    def stdin(self) -> typing.Optional[typing.IO]:  # type: ignore
+    def stdin(self) -> typing.Optional[typing.IO[bytes]]:  # type: ignore
         """Override original NamedTuple with proper typing."""
         return super(SubprocessExecuteAsyncResult, self).stdin
 
     @property
-    def stderr(self) -> typing.Optional[typing.IO]:  # type: ignore
+    def stderr(self) -> typing.Optional[typing.IO[bytes]]:  # type: ignore
         """Override original NamedTuple with proper typing."""
         return super(SubprocessExecuteAsyncResult, self).stderr
 
     @property
-    def stdout(self) -> typing.Optional[typing.IO]:  # type: ignore
+    def stdout(self) -> typing.Optional[typing.IO[bytes]]:  # type: ignore
         """Override original NamedTuple with proper typing."""
         return super(SubprocessExecuteAsyncResult, self).stdout
 
@@ -148,9 +148,9 @@ def close_streams() -> None:
 
         result = exec_result.ExecResult(cmd=cmd_for_log, stdin=stdin, started=async_result.started)
 
-        # noinspection PyNoneFunctionAssignment
+        # noinspection PyNoneFunctionAssignment,PyTypeChecker
         stdout_future: "concurrent.futures.Future[None]" = poll_stdout()
-        # noinspection PyNoneFunctionAssignment
+        # noinspection PyNoneFunctionAssignment,PyTypeChecker
         stderr_future: "concurrent.futures.Future[None]" = poll_stderr()
 
         try:
@@ -218,10 +218,10 @@ def _execute_async(  # pylint: disable=arguments-differ
         :rtype: typing.NamedTuple(
                     'SubprocessExecuteAsyncResult',
                     [
-                        ('interface', subprocess.Popen),
-                        ('stdin', typing.Optional[typing.IO]),
-                        ('stderr', typing.Optional[typing.IO]),
-                        ('stdout', typing.Optional[typing.IO]),
+                        ('interface', subprocess.Popen[bytes]),
+                        ('stdin', typing.Optional[typing.IO[bytes]]),
+                        ('stderr', typing.Optional[typing.IO[bytes]]),
+                        ('stdout', typing.Optional[typing.IO[bytes]]),
                         ("started", datetime.datetime),
                     ]
                 )
@@ -234,7 +234,7 @@ def _execute_async(  # pylint: disable=arguments-differ
         """
         started = datetime.datetime.utcnow()
 
-        process = subprocess.Popen(
+        process: "subprocess.Popen[bytes]" = subprocess.Popen(
             args=[self._prepare_command(cmd=command, chroot_path=chroot_path)],
             stdout=subprocess.PIPE if open_stdout else subprocess.DEVNULL,
             stderr=subprocess.PIPE if open_stderr else subprocess.DEVNULL,
@@ -247,7 +247,7 @@ def _execute_async(  # pylint: disable=arguments-differ
         )
 
         if stdin is None:
-            process_stdin: typing.Optional[typing.IO[typing.Any]] = process.stdin
+            process_stdin: typing.Optional[typing.IO[bytes]] = process.stdin
         else:
             stdin_str: bytes = self._string_bytes_bytearray_as_bytes(stdin)
             try: