How to Capture Stdout Output from a Python Function Call

How to capture stdout output from a Python function call?

Try this context manager:

from io import StringIO 
import sys

class Capturing(list):
    def __enter__(self):
        self._stdout = sys.stdout
        sys.stdout = self._stringio = StringIO()
        return self
    def __exit__(self, *args):
        self.extend(self._stringio.getvalue().splitlines())
        del self._stringio    # free up some memory
        sys.stdout = self._stdout

Usage:

with Capturing() as output:
    do_something(my_object)

output is now a list containing the lines printed by the function call.

Advanced usage:

What may not be obvious is that this can be done more than once and the results concatenated:

with Capturing() as output:
    print('hello world')

print('displays on screen')

with Capturing(output) as output:  # note the constructor argument
    print('hello world2')

print('done')
print('output:', output)

Output:

displays on screen                     
done                                   
output: ['hello world', 'hello world2']

Update: They added redirect_stdout() to contextlib in Python 3.4 (along with redirect_stderr()). So you could use io.StringIO with that to achieve a similar result (though Capturing being a list as well as a context manager is arguably more convenient).

Python - Capture all ouput from function call executing external program

You could give a shot redirecting from sys.__stdout__ in your Capturing class, just to eliminate the possibility that nose or other plugins within nose are messing with you capturing the output. This object contains the original values of stdout at the start of the program, as described in sys docs. This will only work if your executable process already properly redirected its stdout stream to the python interpreter's stream.

But I think it would not make much difference. The key here is that capturing and redirecting sys.stdout in most cases doesn’t affect the standard I/O streams of processes executed by subprocess() family of functions, which I suppose is happening here.

Your best bet is to explicitly capture stdout from your subprocess.Popen call using subprocess.PIPE and communicate() (like here), print it out within the test and let the nose capture it. There is a good chance that subprocess call is done via shell and is outside the stdout which python has control over.

And if you cannot modify this code, you can always monkey patch it ;-)

Capture stdout from a script?

Setting stdout is a reasonable way to do it. Another is to run it as another process:

import subprocess

proc = subprocess.Popen(["python", "-c", "import writer; writer.write()"], stdout=subprocess.PIPE)
out = proc.communicate()[0]
print out.upper()

How to capture prints in real time from function?

You could write your own file-like object that processes lines of text as it sees them. In the simplest case you only need to supply a write method as shown below. The tricky part is knowing when a "print" call is done. print may call stdout.write several times to do a single print operation. In this example, I did processing whenever a newline is seen. This code does not return interim prints but does allow you to intercept the writes to stdout and process them before returning to the function that calls print.

from contextlib import redirect_stdout
import sys

real_stdout_for_test = sys.stdout

class WriteProcessor:

    def __init__(self):
        self.buf = ""

    def write(self, buf):
        # emit on each newline
        while buf:
            try:
                newline_index = buf.index("\n")
            except ValueError:
                # no newline, buffer for next call
                self.buf += buf
                break
            # get data to next newline and combine with any buffered data
            data = self.buf + buf[:newline_index + 1]
            self.buf = ""
            buf = buf[newline_index + 1:]
            # perform complex calculations... or just print with a note.
            real_stdout_for_test.write("fiddled with " + data)
            
with redirect_stdout(WriteProcessor()):
    print("hello there")
    print("a\nprint\nof\nmany\nlines")
    print("goodbye ", end="")
    print("for now")

Access the printed output of a function call

As @JimDeville commented, you can swap stdout:

#!python2.7
import io
import sys

def foo():
    print 'hello, world!'

capture = io.BytesIO()
save,sys.stdout = sys.stdout,capture
foo()
sys.stdout = save
print capture.getvalue()

Output:

hello, world!

Python 3 version uses io.StringIO instead due to stdout expected to be a Unicode stream:

#!python3
import io
import sys

def foo():
    print('hello, world!')

capture = io.StringIO()
save,sys.stdout = sys.stdout,capture
foo()
sys.stdout = save
print(capture.getvalue())

Real time stdout redirect from a python function call to an async method

Thanks for the comment of @HTF I finally managed to solve the problem with janus. I copy the example of the repo, and I modified in order to receive a variable number of messages (because I don't know how many iterations my_heavy_function() will use)

import asyncio
import janus
import time 

def my_heavy_function(sync_q):
    for i in range(10):
        sync_q.put(i)
        time.sleep(1)
    sync_q.put('end')    # is there a more elegant way to do this ?
    sync_q.join()

async def async_coro(async_q):
    while True:
        val = await async_q.get()
        print(f'arrived {val}')
        # send val to UI 
        # await push_message_to_user(val)
        async_q.task_done()
        if val == 'end': 
            break

async def main():
    queue = janus.Queue()
    loop = asyncio.get_running_loop()
    fut = loop.run_in_executor(None, my_heavy_function, queue.sync_q)
    await async_coro(queue.async_q)
    await fut
    queue.close()
    await queue.wait_closed()

asyncio.run(main())

Capturing stdout within the same process in Python

You said that your script "calls a bunch of functions" so I'm assuming that they're python functions accessible from your program. I'm also assuming you're using print to generate the output in all these functions. If that's the case, you can just replace sys.stdout with a StringIO.StringIO which will intercept all the stuff you're writing. Then you can finally call the .getValue method on your StringIO to get everything that has been sent to the output channel. This will also work for external programs using the subprocess module which write to sys.stdout.

This is a cheap way. I'd recommend that you do your output using the logging module. You'll have much more control over how it does it's output and you can control it more easily as well.

Python: capture and save function output

First create an objet that will store your outputs with a write method:

class EmittingStream:
    def __init__(self, path_logs):
        self.path_logs = path_logs

    def write(self, text):
        f = open(self.path_logs, "a")
        f.write(text)
        f.close()

Then associate this class with stdout:

sys.stdout = EmittingStream("my/path/to/logs.txt")

Here a full mini example with a decorator:

def logger(f):
    class EmittingStream:
        def __init__(self, path_logs):
            self.path_logs = path_logs

        def write(self, text):
            f = open(self.path_logs, "a")
            f.write(text)
            f.close()

    def wrapper(*args, **kwargs):
        save_std_out = sys.stdout
        sys.stdout = EmittingStream(os.path.join(dir, "test.txt"))
        res = f()
        sys.stdout = save_std_out
        return res
    return wrapper

@logger
def f():
    print("ok")
 
if __name__ == "__main__":
    f()

Not that you should save the default std_out to restore it after.

Running shell command and capturing the output

In all officially maintained versions of Python, the simplest approach is to use the subprocess.check_output function:

>>> subprocess.check_output(['ls', '-l'])
b'total 0\n-rw-r--r--  1 memyself  staff  0 Mar 14 11:04 files\n'

check_output runs a single program that takes only arguments as input.¹ It returns the result exactly as printed to stdout. If you need to write input to stdin, skip ahead to the run or Popen sections. If you want to execute complex shell commands, see the note on shell=True at the end of this answer.

The check_output function works in all officially maintained versions of Python. But for more recent versions, a more flexible approach is available.

Modern versions of Python (3.5 or higher): run

If you're using Python 3.5+, and do not need backwards compatibility, the new run function is recommended by the official documentation for most tasks. It provides a very general, high-level API for the subprocess module. To capture the output of a program, pass the subprocess.PIPE flag to the stdout keyword argument. Then access the stdout attribute of the returned CompletedProcess object:

>>> import subprocess
>>> result = subprocess.run(['ls', '-l'], stdout=subprocess.PIPE)
>>> result.stdout
b'total 0\n-rw-r--r--  1 memyself  staff  0 Mar 14 11:04 files\n'

The return value is a bytes object, so if you want a proper string, you'll need to decode it. Assuming the called process returns a UTF-8-encoded string:

>>> result.stdout.decode('utf-8')
'total 0\n-rw-r--r--  1 memyself  staff  0 Mar 14 11:04 files\n'

This can all be compressed to a one-liner if desired:

>>> subprocess.run(['ls', '-l'], stdout=subprocess.PIPE).stdout.decode('utf-8')
'total 0\n-rw-r--r--  1 memyself  staff  0 Mar 14 11:04 files\n'

If you want to pass input to the process's stdin, you can pass a bytes object to the input keyword argument:

>>> cmd = ['awk', 'length($0) > 5']
>>> ip = 'foo\nfoofoo\n'.encode('utf-8')
>>> result = subprocess.run(cmd, stdout=subprocess.PIPE, input=ip)
>>> result.stdout.decode('utf-8')
'foofoo\n'

You can capture errors by passing stderr=subprocess.PIPE (capture to result.stderr) or stderr=subprocess.STDOUT (capture to result.stdout along with regular output). If you want run to throw an exception when the process returns a nonzero exit code, you can pass check=True. (Or you can check the returncode attribute of result above.) When security is not a concern, you can also run more complex shell commands by passing shell=True as described at the end of this answer.

Later versions of Python streamline the above further. In Python 3.7+, the above one-liner can be spelled like this:

>>> subprocess.run(['ls', '-l'], capture_output=True, text=True).stdout
'total 0\n-rw-r--r--  1 memyself  staff  0 Mar 14 11:04 files\n'

Using run this way adds just a bit of complexity, compared to the old way of doing things. But now you can do almost anything you need to do with the run function alone.

Older versions of Python (3-3.4): more about check_output

If you are using an older version of Python, or need modest backwards compatibility, you can use the check_output function as briefly described above. It has been available since Python 2.7.

subprocess.check_output(*popenargs, **kwargs)  

It takes takes the same arguments as Popen (see below), and returns a string containing the program's output. The beginning of this answer has a more detailed usage example. In Python 3.5+, check_output is equivalent to executing run with check=True and stdout=PIPE, and returning just the stdout attribute.

You can pass stderr=subprocess.STDOUT to ensure that error messages are included in the returned output. When security is not a concern, you can also run more complex shell commands by passing shell=True as described at the end of this answer.

If you need to pipe from stderr or pass input to the process, check_output won't be up to the task. See the Popen examples below in that case.

Complex applications and legacy versions of Python (2.6 and below): Popen

If you need deep backwards compatibility, or if you need more sophisticated functionality than check_output or run provide, you'll have to work directly with Popen objects, which encapsulate the low-level API for subprocesses.

The Popen constructor accepts either a single command without arguments, or a list containing a command as its first item, followed by any number of arguments, each as a separate item in the list. shlex.split can help parse strings into appropriately formatted lists. Popen objects also accept a host of different arguments for process IO management and low-level configuration.

To send input and capture output, communicate is almost always the preferred method. As in:

output = subprocess.Popen(["mycmd", "myarg"], 
                          stdout=subprocess.PIPE).communicate()[0]

Or

>>> import subprocess
>>> p = subprocess.Popen(['ls', '-a'], stdout=subprocess.PIPE, 
...                                    stderr=subprocess.PIPE)
>>> out, err = p.communicate()
>>> print out
.
..
foo

If you set stdin=PIPE, communicate also allows you to pass data to the process via stdin:

>>> cmd = ['awk', 'length($0) > 5']
>>> p = subprocess.Popen(cmd, stdout=subprocess.PIPE,
...                           stderr=subprocess.PIPE,
...                           stdin=subprocess.PIPE)
>>> out, err = p.communicate('foo\nfoofoo\n')
>>> print out
foofoo

Note Aaron Hall's answer, which indicates that on some systems, you may need to set stdout, stderr, and stdin all to PIPE (or DEVNULL) to get communicate to work at all.

In some rare cases, you may need complex, real-time output capturing. Vartec's answer suggests a way forward, but methods other than communicate are prone to deadlocks if not used carefully.

As with all the above functions, when security is not a concern, you can run more complex shell commands by passing shell=True.

Notes

1. Running shell commands: the shell=True argument

Normally, each call to run, check_output, or the Popen constructor executes a single program. That means no fancy bash-style pipes. If you want to run complex shell commands, you can pass shell=True, which all three functions support. For example:

>>> subprocess.check_output('cat books/* | wc', shell=True, text=True)
' 1299377 17005208 101299376\n'

However, doing this raises security concerns. If you're doing anything more than light scripting, you might be better off calling each process separately, and passing the output from each as an input to the next, via

run(cmd, [stdout=etc...], input=other_output)

Or

Popen(cmd, [stdout=etc...]).communicate(other_output)

The temptation to directly connect pipes is strong; resist it. Otherwise, you'll likely see deadlocks or have to do hacky things like this.

---

Related Topics