add parallel stdout/stderr streaming option

Author: AAriam

pkg/pyproject.toml

@@ -15,7 +15,7 @@ namespaces = true
 
 # ----------------------------------------- Project Metadata -------------------------------------
 [project]
-version = "0.1.0"
+version = "0.2.0"
 name = "PyShellMan"
 requires-python = ">=3.10"
 dependencies = [

pkg/src/pyshellman/shell.py

@@ -4,6 +4,8 @@
 from types import SimpleNamespace as _SimpleNamespace
 import subprocess as _subprocess
 from pathlib import Path as _Path
+import threading as _threading
+import sys as _sys
 
 from pyshellman import exception as _exception
 from pyshellman.output import ShellOutput as _ShellOutput
@@ -15,8 +17,11 @@ class Runner:
 
     def __init__(
         self,
+        *,
         pre_command: list[str] | None = None,
         cwd: str | _Path | None = None,
+        stream_stdout: bool = False,
+        stream_stderr: bool = False,
         raise_execution: bool = True,
         raise_exit_code: bool = True,
         raise_stderr: bool = False,
@@ -31,6 +36,8 @@ def __init__(
     ):
         self.pre_command = pre_command or []
         self.cwd = _Path(cwd).resolve() if cwd else None
+        self.stream_stdout = stream_stdout
+        self.stream_stderr = stream_stderr
         self.raise_execution = raise_execution
         self.raise_exit_code = raise_exit_code
         self.raise_stderr = raise_stderr
@@ -47,7 +54,10 @@ def __init__(
     def run(
         self,
         command: list[str],
+        *,
         cwd: str | _Path | None = None,
+        stream_stdout: bool | None = None,
+        stream_stderr: bool | None = None,
         raise_execution: bool | None = None,
         raise_exit_code: bool | None = None,
         raise_stderr: bool | None = None,
@@ -62,16 +72,17 @@ def run(
         args = self._get_run_args(locals())
         command = self.pre_command + command
         cwd = _Path(args.cwd).resolve() if args.cwd else None
-        try:
-            process = _subprocess.run(command, text=args.text_output, cwd=args.cwd, capture_output=True)
-        except FileNotFoundError:
-            stdout = None
-            stderr = None
-            code = None
+
+        if args.stream_stdout or args.stream_stderr:
+            func = self._run_stream
+            kwargs = {
+                "stream_stdout": args.stream_stdout,
+                "stream_stderr": args.stream_stderr,
+            }
         else:
-            stdout = (process.stdout.strip() if args.text_output else process.stdout) or None
-            stderr = (process.stderr.strip() if args.text_output else process.stderr) or None
-            code = process.returncode
+            func = self._run_nostream
+            kwargs = {}
+        stdout, stderr, code = func(command, args, **kwargs)
         output = _ShellOutput(
             title=args.log_title,
             command=command,
@@ -110,10 +121,93 @@ def _get_run_args(self, args: dict) -> _SimpleNamespace:
         }
         return _SimpleNamespace(**final_args)
 
+    @staticmethod
+    def _run_nostream(
+        command: list[str],
+        args: _SimpleNamespace
+    ) -> tuple[str | bytes | None, str | bytes | None, int | None]:
+        try:
+            process = _subprocess.run(command, text=args.text_output, cwd=args.cwd, capture_output=True)
+        except FileNotFoundError:
+            return None, None, None
+        stdout = (process.stdout.strip() if args.text_output else process.stdout) or None
+        stderr = (process.stderr.strip() if args.text_output else process.stderr) or None
+        code = process.returncode
+        return stdout, stderr, code
+
+    @staticmethod
+    def _run_stream(
+        command: list[str],
+        args: _SimpleNamespace,
+        stream_stdout: bool,
+        stream_stderr: bool,
+    ) -> tuple[str | bytes | None, str | bytes | None, int | None]:
+        """Run a subprocess while printing and capturing stdout/stderr."""
+        try:
+            process = _subprocess.Popen(
+                command,
+                text=args.text_output,
+                cwd=args.cwd,
+                stdout=_subprocess.PIPE,
+                stderr=_subprocess.PIPE,
+                bufsize=1 if args.text_output else 0,
+            )
+        except FileNotFoundError:
+            return None, None, None
+
+        stdout_chunks: list[str] | list[bytes] = []
+        stderr_chunks: list[str] | list[bytes] = []
+
+        def read_stream(stream, chunks, live: bool):
+            if args.text_output:
+                for line in iter(stream.readline, ''):
+                    chunks.append(line)
+                    if live:
+                        print(line, end='', flush=True)
+            else:
+                # Read in binary chunks; choose a reasonable block size (e.g., 4096)
+                while True:
+                    chunk = stream.read(4096)
+                    if not chunk:
+                        break
+                    if live:
+                        # Write bytes directly to stdout.buffer (bypass text encoding)
+                        _sys.stdout.buffer.write(chunk)
+                        _sys.stdout.buffer.flush()
+                    chunks.append(chunk)
+            stream.close()
+
+        threads = [
+            _threading.Thread(
+                target=read_stream,
+                args=(process.stdout, stdout_chunks, stream_stdout),
+            ),
+            _threading.Thread(
+                target=read_stream,
+                args=(process.stderr, stderr_chunks, stream_stderr),
+            ),
+        ]
+
+        for t in threads:
+            t.start()
+        process.wait()
+        for t in threads:
+            t.join()
+        if args.text_output:
+            stdout = ''.join(stdout_chunks)
+            stderr = ''.join(stderr_chunks)
+        else:
+            stdout = b''.join(stdout_chunks)
+            stderr = b''.join(stderr_chunks)
+        return stdout, stderr, process.returncode
+
 
 def run(
     command: list[str],
+    *,
     cwd: str | _Path | None = None,
+    stream_stdout: bool = False,
+    stream_stderr: bool = False,
     raise_execution: bool = True,
     raise_exit_code: bool = True,
     raise_stderr: bool = False,
@@ -128,6 +222,8 @@ def run(
 ) -> _ShellOutput:
     return Runner(
         cwd=cwd,
+        stream_stdout=stream_stdout,
+        stream_stderr=stream_stderr,
         raise_execution=raise_execution,
         raise_exit_code=raise_exit_code,
         raise_stderr=raise_stderr,