Allow parallel run of commands (#83)

penguinolog · web-flow · commit c01eda002b62 · 2018-09-21T10:11:26.000+02:00
* all execute methods can be started in parallel
  if required - use `lock` attribute

* Change `execute_async` signature and use typed NamedTuple.
  Types are exported for typing purposes only.
diff --git a/doc/source/SSHClient.rst b/doc/source/SSHClient.rst
@@ -138,11 +138,12 @@ API: SSHClient and SSHAuth.
         :param log_mask_re: regex lookup rule to mask command for logger.
                             all MATCHED groups will be replaced by '<*masked*>'
         :type log_mask_re: typing.Optional[str]
-        :rtype: ``typing.Tuple[paramiko.Channel, paramiko.ChannelFile, typing.Optional[paramiko.ChannelFile], typing.Optional[paramiko.ChannelFile]]``
+        :rtype: SshExecuteAsyncResult
 
         .. versionchanged:: 1.2.0 open_stdout and open_stderr flags
         .. versionchanged:: 1.2.0 stdin data
         .. versionchanged:: 1.2.0 get_pty moved to `**kwargs`
+        .. versionchanged:: 2.1.0 Use typed NamedTuple as result
 
     .. py:method:: execute(command, verbose=False, timeout=1*60*60, **kwargs)
 
@@ -388,3 +389,24 @@ API: SSHClient and SSHAuth.
         :param log: Log on generic connection failure
         :type log: ``bool``
         :raises paramiko.AuthenticationException: Authentication failed.
+
+
+.. py:class:: SshExecuteAsyncResult
+
+    Typed NamedTuple
+
+    .. py:attribute:: interface
+
+        ``paramiko.Channel``
+
+    .. py:attribute:: stdin
+
+        ``paramiko.ChannelFile``
+
+    .. py:attribute:: stderr
+
+        ``typing.Optional[paramiko.ChannelFile]``
+
+    .. py:attribute:: stdout
+
+        ``typing.Optional[paramiko.ChannelFile]``
diff --git a/doc/source/Subprocess.rst b/doc/source/Subprocess.rst
@@ -56,9 +56,11 @@ API: Subprocess
         :param log_mask_re: regex lookup rule to mask command for logger.
                             all MATCHED groups will be replaced by '<*masked*>'
         :type log_mask_re: ``typing.Optional[str]``
-        :rtype: ``typing.Tuple[subprocess.Popen, None, typing.Optional[typing.IO], typing.Optional[typing.IO], ]``
+        :rtype: SubprocessExecuteAsyncResult
+        :raises OSError: impossible to process STDIN
 
         .. versionadded:: 1.2.0
+        .. versionchanged:: 2.1.0 Use typed NamedTuple as result
 
     .. py:method:: execute(command, verbose=False, timeout=1*60*60, **kwargs)
 
@@ -124,3 +126,24 @@ API: Subprocess
 
         .. versionchanged:: 1.1.0 make method
         .. versionchanged:: 1.2.0 default timeout 1 hour
+
+
+.. py:class:: SubprocessExecuteAsyncResult
+
+    Typed NamedTuple
+
+    .. py:attribute:: interface
+
+        ``subprocess.Popen``
+
+    .. py:attribute:: stdin
+
+        ``typing.Optional[typing.IO]``
+
+    .. py:attribute:: stderr
+
+        ``typing.Optional[typing.IO]``
+
+    .. py:attribute:: stdout
+
+        ``typing.Optional[typing.IO]``
diff --git a/exec_helpers/__init__.py b/exec_helpers/__init__.py
@@ -28,8 +28,9 @@
 from .exec_result import ExecResult
 from .api import ExecHelper
 from .ssh_auth import SSHAuth
+from ._ssh_client_base import SshExecuteAsyncResult
 from .ssh_client import SSHClient
-from .subprocess_runner import Subprocess  # nosec  # Expected
+from .subprocess_runner import Subprocess, SubprocessExecuteAsyncResult  # nosec  # Expected
 
 __all__ = (
     'ExecHelperError',
@@ -40,8 +41,10 @@
     'ExecHelperTimeoutError',
     'ExecHelper',
     'SSHClient',
+    'SshExecuteAsyncResult',
     'SSHAuth',
     'Subprocess',
+    'SubprocessExecuteAsyncResult',
     'ExitCodes',
     'ExecResult',
 )
diff --git a/exec_helpers/_ssh_client_base.py b/exec_helpers/_ssh_client_base.py
@@ -16,6 +16,11 @@
 
 """SSH client helper based on Paramiko. Base class."""
 
+__all__ = (
+    'SSHClientBase',
+    'SshExecuteAsyncResult',
+)
+
 import abc
 import base64
 import collections
@@ -43,18 +48,33 @@
 from exec_helpers import ssh_auth
 from exec_helpers import _log_templates
 
-__all__ = ('SSHClientBase', )
-
 logging.getLogger('paramiko').setLevel(logging.WARNING)
 logging.getLogger('iso8601').setLevel(logging.WARNING)
 
 
-_type_execute_async = typing.Tuple[
-    paramiko.Channel,
-    paramiko.ChannelFile,
-    typing.Optional[paramiko.ChannelFile],
-    typing.Optional[paramiko.ChannelFile]
-]
+class SshExecuteAsyncResult(api.ExecuteAsyncResult):
+    """Override original NamedTuple with proper typing."""
+
+    @property
+    def interface(self) -> paramiko.Channel:
+        """Override original NamedTuple with proper typing."""
+        return super(SshExecuteAsyncResult, self).interface
+
+    @property
+    def stdin(self) -> paramiko.ChannelFile:  # type: ignore
+        """Override original NamedTuple with proper typing."""
+        return super(SshExecuteAsyncResult, self).stdin
+
+    @property
+    def stderr(self) -> typing.Optional[paramiko.ChannelFile]:  # type: ignore
+        """Override original NamedTuple with proper typing."""
+        return super(SshExecuteAsyncResult, self).stderr
+
+    @property
+    def stdout(self) -> typing.Optional[paramiko.ChannelFile]:  # type: ignore
+        """Override original NamedTuple with proper typing."""
+        return super(SshExecuteAsyncResult, self).stdout
+
 
 CPYTHON = 'CPython' == platform.python_implementation()
 
@@ -584,7 +604,7 @@ def execute_async(
         verbose: bool = False,
         log_mask_re: typing.Optional[str] = None,
         **kwargs: typing.Any
-    ) -> _type_execute_async:
+    ) -> SshExecuteAsyncResult:
         """Execute command in async mode and return channel with IO objects.
 
         :param command: Command for execution
@@ -603,16 +623,20 @@ def execute_async(
         :param kwargs: additional parameters for call.
         :type kwargs: typing.Any
         :return: Tuple with control interface and file-like objects for STDIN/STDERR/STDOUT
-        :rtype: typing.Tuple[
-            paramiko.Channel,
-            paramiko.ChannelFile,
-            typing.Optional[paramiko.ChannelFile],
-            typing.Optional[paramiko.ChannelFile],
-        ]
+        :rtype: typing.NamedTuple(
+                    'SshExecuteAsyncResult',
+                    [
+                        ('interface', paramiko.Channel),
+                        ('stdin', paramiko.ChannelFile),
+                        ('stderr', typing.Optional[paramiko.ChannelFile]),
+                        ('stdout', typing.Optional[paramiko.ChannelFile]),
+                    ]
+                )
 
         .. versionchanged:: 1.2.0 open_stdout and open_stderr flags
         .. versionchanged:: 1.2.0 stdin data
         .. versionchanged:: 1.2.0 get_pty moved to `**kwargs`
+        .. versionchanged:: 2.1.0 Use typed NamedTuple as result
         """
         cmd_for_log = self._mask_command(
             cmd=command,
@@ -657,7 +681,7 @@ def execute_async(
             else:
                 self.logger.warning('STDIN Send failed: closed channel')
 
-        return chan, _stdin, stderr, stdout
+        return SshExecuteAsyncResult(chan, _stdin, stderr, stdout)
 
     def _exec_command(
         self,
@@ -887,18 +911,10 @@ def execute_together(
         @threaded.threadpooled  # type: ignore
         def get_result(remote: 'SSHClientBase') -> exec_result.ExecResult:
             """Get result from remote call."""
-            (
-                chan,
-                _,
-                stderr,
-                stdout,
-            ) = remote.execute_async(
-                command,
-                **kwargs
-            )  # type: _type_execute_async
+            async_result = remote.execute_async(command, **kwargs)  # type: SshExecuteAsyncResult
 
-            chan.status_event.wait(timeout)
-            exit_code = chan.recv_exit_status()
+            async_result.interface.status_event.wait(timeout)
+            exit_code = async_result.interface.recv_exit_status()
 
             # pylint: disable=protected-access
             cmd_for_log = remote._mask_command(
@@ -908,11 +924,11 @@ def get_result(remote: 'SSHClientBase') -> exec_result.ExecResult:
             # pylint: enable=protected-access
 
             result = exec_result.ExecResult(cmd=cmd_for_log)
-            result.read_stdout(src=stdout)
-            result.read_stderr(src=stderr)
+            result.read_stdout(src=async_result.stdout)
+            result.read_stderr(src=async_result.stderr)
             result.exit_code = exit_code
 
-            chan.close()
+            async_result.interface.close()
             return result
 
         expected = expected or [proc_enums.ExitCodes.EX_OK]
diff --git a/exec_helpers/api.py b/exec_helpers/api.py
@@ -19,6 +19,11 @@
 .. versionchanged:: 1.3.5 make API public to use as interface
 """
 
+__all__ = (
+    'ExecHelper',
+    'ExecuteAsyncResult',
+)
+
 import abc
 import logging
 import re
@@ -31,6 +36,17 @@
 from exec_helpers import proc_enums
 
 
+ExecuteAsyncResult = typing.NamedTuple(
+    'ExecuteAsyncResult',
+    [
+        ('interface', typing.Any),
+        ('stdin', typing.Optional[typing.Any]),
+        ('stderr', typing.Optional[typing.Any]),
+        ('stdout', typing.Optional[typing.Any]),
+    ]
+)
+
+
 class ExecHelper(metaclass=abc.ABCMeta):
     """ExecHelper global API."""
 
@@ -142,7 +158,7 @@ def execute_async(
         verbose: bool = False,
         log_mask_re: typing.Optional[str] = None,
         **kwargs: typing.Any
-    ) -> typing.Tuple[typing.Any, typing.Any, typing.Any, typing.Any]:
+    ) -> ExecuteAsyncResult:
         """Execute command in async mode and return remote interface with IO objects.
 
         :param command: Command for execution
@@ -160,11 +176,20 @@ def execute_async(
         :type log_mask_re: typing.Optional[str]
         :param kwargs: additional parameters for call.
         :type kwargs: typing.Any
-        :return: Tuple with control interface and file-like objects for STDIN/STDERR/STDOUT
-        :rtype: typing.Tuple[typing.Any, typing.Any, typing.Any, typing.Any]
+        :return: NamedTuple with control interface and file-like objects for STDIN/STDERR/STDOUT
+        :rtype: typing.NamedTuple(
+                    'ExecuteAsyncResult',
+                    [
+                        ('interface', typing.Any),
+                        ('stdin', typing.Optional[typing.Any]),
+                        ('stderr', typing.Optional[typing.Any]),
+                        ('stdout', typing.Optional[typing.Any]),
+                    ]
+                )
 
         .. versionchanged:: 1.2.0 open_stdout and open_stderr flags
         .. versionchanged:: 1.2.0 stdin data
+        .. versionchanged:: 2.1.0 Use typed NamedTuple as result
         """
         raise NotImplementedError  # pragma: no cover
 
@@ -229,34 +254,25 @@ def execute(
         :raises ExecHelperTimeoutError: Timeout exceeded
 
         .. versionchanged:: 1.2.0 default timeout 1 hour
+        .. versionchanged:: 2.1.0 Allow parallel calls
         """
-        with self.lock:
-            (
-                iface,
-                _,
-                stderr,
-                stdout,
-            ) = self.execute_async(
-                command,
-                verbose=verbose,
-                **kwargs
-            )
-
-            result = self._exec_command(
-                command=command,
-                interface=iface,
-                stdout=stdout,
-                stderr=stderr,
-                timeout=timeout,
-                verbose=verbose,
-                **kwargs
-            )
-            message = "Command {result.cmd!r} exit code: {result.exit_code!s}".format(result=result)
-            self.logger.log(  # type: ignore
-                level=logging.INFO if verbose else logging.DEBUG,
-                msg=message
-            )
-            return result
+        async_result = self.execute_async(command, verbose=verbose, **kwargs)  # type: ExecuteAsyncResult
+
+        result = self._exec_command(
+            command=command,
+            interface=async_result.interface,
+            stdout=async_result.stdout,
+            stderr=async_result.stderr,
+            timeout=timeout,
+            verbose=verbose,
+            **kwargs
+        )
+        message = "Command {result.cmd!r} exit code: {result.exit_code!s}".format(result=result)
+        self.logger.log(  # type: ignore
+            level=logging.INFO if verbose else logging.DEBUG,
+            msg=message
+        )
+        return result
 
     def check_call(
         self,
diff --git a/exec_helpers/subprocess_runner.py b/exec_helpers/subprocess_runner.py
diff --git a/test/test_ssh_client.py b/test/test_ssh_client.py
