live-tee-and-capture

Run a command and tee its stdout/stderr in real time to your terminal while also capturing them.

• Outputs are relayed in real time with no artificial buffering (a dedicated thread drains each pipe as data arrives) - ideal for real-time logs and progress bars.
• You can enable/disable teeing stdout/stderr independently.
• Optional timeout: the command's direct process is killed and TimeoutExpired is raised if it does not exit in time.
• File descriptors and reader threads are always cleaned up, even if launching fails or the command times out.
• Descendant processes are not killed automatically; inherited output pipes cannot delay completion after the direct process exits.

Usage

Example

from live_tee_and_capture import live_tee_and_capture

exit_code, stdout_bytes, stderr_bytes = live_tee_and_capture(
    ['ls', '-la'],
    tee_stdout=True,
    tee_stderr=True,
)

print('Exit code:', exit_code)
print('Captured stdout:', stdout_bytes.decode())
print('Captured stderr:', stderr_bytes.decode())

Function Signature

def live_tee_and_capture(
    command: Sequence[Text],
    tee_stdout: bool = True,
    tee_stderr: bool = True,
    timeout: Optional[float] = None,
) -> Tuple[int, bytearray, bytearray]:
    ...

If timeout (in seconds) is given and the command does not exit within that time, its direct process is killed and a TimeoutExpired exception is raised. Descendant processes are not terminated automatically:

from live_tee_and_capture import live_tee_and_capture, TimeoutExpired

try:
    live_tee_and_capture(['sleep', '30'], timeout=5)
except TimeoutExpired as e:
    print('Timed out:', e.command, e.timeout)
    # Output captured before the command was killed:
    print('Partial stdout:', e.stdout.decode())
    print('Partial stderr:', e.stderr.decode())

Contributing

Contributions are welcome! Please submit pull requests or open issues on the GitHub repository.

License

This project is licensed under the MIT License.