syscmd.py

#!/usr/bin/env python3

# File: popen_out_gen.py
# SPDX-License-Identifier: Unlicense

# This is free and unencumbered software released into the public domain.
#
# Anyone is free to copy, modify, publish, use, compile, sell, or
# distribute this software, either in source code form or as a compiled
# binary, for any purpose, commercial or non-commercial, and by any
# means.

# Tectonics:
#   $ black syscmd.py

import shutil
import subprocess
import sys


def popen_out_gen(cmd, args, **params):
    """Executes the command cmd with args via subprocess.Popen().

    The params are passed to subprocess.Popen. By default, the params
    passed to Popen are:
      * stdout=subprocess.PIPE
      * text=True

    Yields each line of the subprocess output to the caller.
    The line ending "\n" is *not* removed.

    If shell is True then the args are combined into a single
    platform-specific string and run within a system shell.

    Once the subprocess stops writing to output, closes
    the ouptut stream and waits for the subprocess to end.

    If the subprocess returned a non-zero return code,
    then raises a subprocess.CalledProcessError exception.
    """
    # Prepend the (fully qualified) cmd to args.
    # This modifies the args parameter
    if shutil.which(cmd):
        args.insert(0, shutil.which(cmd))
    else:
        args.insert(0, cmd)

    # Run the given command plus args as a subprocess,
    # yielding each output line to the caller
    kwargs = {
        "stdout": subprocess.PIPE,
        "text": True,
    }
    kwargs.update(**params)

    # Combine args if running via a shell
    # CAVEAT: running as a shell process does not work reliably
    if kwargs.get("shell", None):
        args = " ".join(args)
    proc = subprocess.Popen(args, **kwargs)

    out_readable = hasattr(proc.stdout, "readline")
    err_readable = hasattr(proc.stderr, "readline")
    while out_readable or err_readable:
        if out_readable:
            line = proc.stdout.readline()
            yield line
            if line == "":
                out_readable = False
        if err_readable:
            line = proc.stderr.readline()
            yield line
            if line == "":
                err_readable = False

    # Wait for the subprocess to end,
    return_code = proc.wait()

    # Finally, raise a CalledProcessError if the return code != 0
    if return_code != 0:
        raise subprocess.CalledProcessError(return_code, args)