Run subprocess and print output to logging

You could try to pass the pipe directly without buffering the whole subprocess output in memory:

from subprocess import Popen, PIPE, STDOUT

process = Popen(command_line_args, stdout=PIPE, stderr=STDOUT)
with process.stdout:
    log_subprocess_output(process.stdout)
exitcode = process.wait() # 0 means success

where log_subprocess_output() could look like:

def log_subprocess_output(pipe):
    for line in iter(pipe.readline, b''): # b'\n'-separated lines
        logging.info('got line from subprocess: %r', line)

I was trying to achieve the same on check_call and check_ouput. I found this solution to be working.

import logging
import threading
import os
import subprocess

logging.basicConfig(format='%(levelname)s:%(message)s', level=logging.INFO)

class LogPipe(threading.Thread):

    def __init__(self, level):
        """Setup the object with a logger and a loglevel
        and start the thread
        """
        threading.Thread.__init__(self)
        self.daemon = False
        self.level = level
        self.fdRead, self.fdWrite = os.pipe()
        self.pipeReader = os.fdopen(self.fdRead)
        self.start()

    def fileno(self):
        """Return the write file descriptor of the pipe"""
        return self.fdWrite

    def run(self):
        """Run the thread, logging everything."""
        for line in iter(self.pipeReader.readline, ''):
            logging.log(self.level, line.strip('\n'))

        self.pipeReader.close()

    def close(self):
        """Close the write end of the pipe."""
        os.close(self.fdWrite)

   def write(self):
       """If your code has something like sys.stdout.write"""
       logging.log(self.level, message)

   def flush(self):
       """If you code has something like this sys.stdout.flush"""
       pass

After implementing it, I performed the below steps:

try:
    # It works on multiple handlers as well
    logging.basicConfig(handlers=[logging.FileHandler(log_file), logging.StreamHandler()])
    sys.stdout = LogPipe(logging.INFO)
    sys.stderr = LogPipe(logging.ERROR)
...
    subprocess.check_call(subprocess_cmd, stdout=sys.stdout, stderr=sys.stderr)
    export_output = subprocess.check_output(subprocess_cmd, stderr=sys.stderr)
...
finally:
    sys.stdout.close()
    sys.stderr.close()
    # It is neccessary to close the file handlers properly.
    sys.stdout = sys.__stdout__
    sys.stderr = sys.__stderr__
    logging.shutdown()
    os.remove(log_file)

This worked for me:

from subprocess import Popen, PIPE, STDOUT

command = f"shell command with arguments"
process = Popen(command, shell=True, stdout=PIPE, stderr=STDOUT)

with process.stdout:
    for line in iter(process.stdout.readline, b''):
        print(line.decode("utf-8").strip())

With exception handling:

from subprocess import Popen, PIPE, STDOUT, CalledProcessError

command = f"shell command with arguments"
process = Popen(command, shell=True, stdout=PIPE, stderr=STDOUT)

with process.stdout:
    try:
        for line in iter(process.stdout.readline, b''):
            print(line.decode("utf-8").strip())
            
    except CalledProcessError as e:
        print(f"{str(e)}")

Handling the live output stream of a command can be achieved by iterating over stdout as the subprocess.Popen runs.

This implementation:

• uses a with-statement such that standard file descriptors are closed, and the process is waited for
• propagates keyword arguments to the subprocess constructor
• defaults to text=True to automatically decode bytestrings into strings
• raises a CalledProcessError upon failure if check=True like subprocess.run does
• returns a CompletedProcess upon success like subprocess.run does
• redirects stderr to stdout (to handle them separately see my multithreaded answer)

import logging
from subprocess import PIPE, STDOUT, CalledProcessError, CompletedProcess, Popen


def stream_command(
    args,
    *,
    stdout_handler=logging.info,  # for a separate stderr_handler (using multithreading), see https://stackoverflow.com/a/76634163/5511061
    check=True,
    text=True,
    **kwargs,
):
    """Mimic subprocess.run, while processing the command output in real time ref https://stackoverflow.com/a/76626021/5511061."""
    with Popen(args, text=text, stdout=PIPE, stderr=STDOUT, **kwargs) as process:
        for line in process.stdout:
            stdout_handler(line[:-1])
    retcode = process.poll()
    if check and retcode:
        raise CalledProcessError(retcode, process.args)
    return CompletedProcess(process.args, retcode)

Logging to a file then becomes as simple as setting up logging:

logging.basicConfig(
    level=logging.INFO,
    filename="./capture.log",
    filemode="w",
    encoding="utf-8",
)
logging.info("test from python")
stream_command(["echo", "test from subprocess"])

With the resulting file:

$ cat ./capture.log
INFO:root:test from python
INFO:root:test from subprocess

The behaviour can be tweaked to preference (print instead of logging.info, or both, etc.):

stream_command(["echo", "test"])
# INFO:root:test
stream_command("cat ./nonexist", shell=True, check=False)
# ERROR:root:cat: ./nonexist: No such file or directory
stream_command(["echo", "test"], stdout_handler=print)
# test
stdout_lines = []
def handler(line):
    print(line)
    logging.info(line)
    stdout_lines.append(line)
stream_command(["echo", "test"], stdout_handler=handler)
print(stdout_lines)
# test
# INFO:root:test
# ['test']