Variant of subprocess.call that accepts a logger instead of stdout/stderr #python

logging_subprocess.py

import subprocess
import select
from logging import DEBUG, ERROR


def call(popenargs, logger, stdout_log_level=DEBUG, stderr_log_level=ERROR, **kwargs):
    """
    Variant of subprocess.call that accepts a logger instead of stdout/stderr,
    and logs stdout messages via logger.debug and stderr messages via
    logger.error.
    """
    child = subprocess.Popen(popenargs, stdout=subprocess.PIPE,
                             stderr=subprocess.PIPE, **kwargs)

    log_level = {child.stdout: stdout_log_level,
                 child.stderr: stderr_log_level}

    def check_io():
        ready_to_read = select.select([child.stdout, child.stderr], [], [], 1000)[0]
        for io in ready_to_read:
            line = io.readline()
            logger.log(log_level[io], line[:-1])

    # keep checking stdout/stderr until the child exits
    while child.poll() is None:
        check_io()

    check_io()  # check again to catch anything after the process exits

    return child.wait()


# tests, plunked in here for convenience

import sys
import unittest2
import logging_subprocess
import logging
from StringIO import StringIO


class LoggingSubprocessTest(unittest2.TestCase):
    def setUp(self):
        self.buffer = StringIO()
        self.logger = logging.getLogger('logging_subprocess_test')
        self.logger.setLevel(logging.DEBUG)
        self.logHandler = logging.StreamHandler(self.buffer)
        formatter = logging.Formatter("%(levelname)s-%(message)s")
        self.logHandler.setFormatter(formatter)
        self.logger.addHandler(self.logHandler)

    def test_log_stdout(self):
        logging_subprocess.call([sys.executable, "-c",
                                "print 'foo'"], self.logger)
        self.assertIn('DEBUG-foo', self.buffer.getvalue())

    def test_log_stderr(self):
        logging_subprocess.call([sys.executable, "-c",
                                'import sys; sys.stderr.write("foo\\n")'],
                                self.logger)
        self.assertIn('ERROR-foo', self.buffer.getvalue())

    def test_custom_stdout_log_level(self):
        logging_subprocess.call([sys.executable, "-c",
                                "print 'foo'"], self.logger,
                                stdout_log_level=logging.INFO)
        self.assertIn('INFO-foo', self.buffer.getvalue())

    def test_custom_stderr_log_level(self):
        logging_subprocess.call([sys.executable, "-c",
                                'import sys; sys.stderr.write("foo\\n")'],
                                self.logger,
                                stderr_log_level=logging.WARNING)
        self.assertIn('WARNING-foo', self.buffer.getvalue())

if __name__ == "__main__":
    unittest2.main()

This won't work on Windows because select() only takes a socket.

I couldn't get this to work on Python 3, I created https://gist.github.com/jaketame/3ed43d1c52e9abccd742b1792c449079 and seems to suit my needs.

There is an issue when a single event produces multiple lines of output, only the first line will be logged. For example if calling with popenargs = ["python", "-c", "raise ValueError()"] you will only get the first line Traceback (most recent call last): output.

This can be fixed by called readlines() rather than readline().
That is change

if rfd == child.stdout.fileno():
    line = child.stdout.readline()
    if len(line) > 0:
        logger.log(stdout_log_level, line[:-1])
if rfd == child.stderr.fileno():
    line = child.stderr.readline()
    if len(line) > 0:
        logger.log(stderr_log_level, line[:-1])

to

if rfd == child.stdout.fileno():
    lines = child.stdout.readlines()
    for line in lines:
        logger.log(stdout_log_level, line[:-1])
if rfd == child.stderr.fileno():
    lines = child.stderr.readlines()
    for line in lines:
        logger.log(stderr_log_level, line[:-1])

This is a great help, even 12 years later, thanks. Here's an update of the tests to work with Python 3 and using @coreydexter modifications for multiple output lines:

import sys
import unittest
from stash_measurement.utils import logging_subprocess
import logging
from io import StringIO


class LoggingSubprocessTest(unittest.TestCase):
    def setUp(self):
        self.buffer = StringIO()
        self.logger = logging.getLogger('logging_subprocess_test')
        self.logger.setLevel(logging.DEBUG)
        self.logHandler = logging.StreamHandler(self.buffer)
        formatter = logging.Formatter("%(levelname)s-%(message)s")
        self.logHandler.setFormatter(formatter)
        self.logger.addHandler(self.logHandler)

    def test_log_stdout(self):
        logging_subprocess.call([sys.executable, "-c",
                                "print('foo')"], self.logger)
        self.assertIn("DEBUG-b'foo'\n", self.buffer.getvalue())

    def test_log_stderr(self):
        logging_subprocess.call([sys.executable, "-c",
                                'import sys; sys.stderr.write("foo\\n")'],
                                self.logger)
        self.assertIn("ERROR-b'foo'\n", self.buffer.getvalue())

    def test_custom_stdout_log_level(self):
        logging_subprocess.call([sys.executable, "-c",
                                "print('foo')"], self.logger,
                                stdout_log_level=logging.INFO)
        self.assertIn("INFO-b'foo'\n", self.buffer.getvalue())

    def test_custom_stderr_log_level(self):
        logging_subprocess.call([sys.executable, "-c",
                                'import sys; sys.stderr.write("foo\\n")'],
                                self.logger,
                                stderr_log_level=logging.WARNING)
        self.assertIn("WARNING-b'foo'\n", self.buffer.getvalue())

Hi 👋

I've done a rewrite from scratch:

• Using py 3.7+ text kwarg so it works on Windows special chars as well
• Using two threads in order to handle stdout and stderr separately i.e. logging.INFO and logging.ERROR

https://gist.github.com/ddelange/6517e3267fb74eeee804e3b1490b1c1d