¿Cómo capturar la salida stdout de una llamada a una función de Python?

Estoy usando una biblioteca de Python que hace algo con un objeto.

do_something(my_object)

y lo cambia. Mientras lo hace, imprime algunas estadísticas en stdout, y me gustaría controlar esta información. La solución adecuada sería cambiar do_something()para devolver la información relevante,

out = do_something(my_object)

pero pasará un tiempo antes de que los desarrolladores do_something()lleguen a este tema. Como solución alternativa, pensé en analizar todo lo que se do_something()escribe en stdout.

¿Cómo puedo capturar la salida estándar entre dos puntos en el código, por ejemplo,

start_capturing()
do_something(my_object)
out = end_capturing()

?

Prueba este administrador de contexto:

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

Uso:

with Capturing() as output:
    do_something(my_object)

output es ahora una lista que contiene las líneas impresas por la llamada a la función.

Uso avanzado:

Lo que puede no ser obvio es que esto se puede hacer más de una vez y los resultados concatenados:

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)

Salida:

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

Actualización : Se añaden redirect_stdout()a contextliben Python 3.4 (junto con redirect_stderr()). Así que podría usarlo io.StringIOpara lograr un resultado similar (aunque Capturingpodría decirse que ser una lista y un administrador de contexto es más conveniente).

En python> = 3.4, contextlib contiene un redirect_stdoutdecorador. Se puede usar para responder a su pregunta de la siguiente manera:

import io
from contextlib import redirect_stdout

f = io.StringIO()
with redirect_stdout(f):
    do_something(my_object)
out = f.getvalue()

De los documentos :

Administrador de contexto para redirigir temporalmente sys.stdout a otro archivo u objeto similar a un archivo.

Esta herramienta agrega flexibilidad a las funciones o clases existentes cuya salida está cableada a stdout.

Por ejemplo, la salida de help () normalmente se envía a sys.stdout. Puede capturar esa salida en una cadena redirigiendo la salida a un objeto io.StringIO:

  f = io.StringIO() 
  with redirect_stdout(f):
      help(pow) 
  s = f.getvalue()

Para enviar la salida de help () a un archivo en el disco, redirija la salida a un archivo normal:

 with open('help.txt', 'w') as f:
     with redirect_stdout(f):
         help(pow)

Para enviar la salida de help () a sys.stderr:

with redirect_stdout(sys.stderr):
    help(pow)

Tenga en cuenta que el efecto secundario global en sys.stdout significa que este administrador de contexto no es adecuado para su uso en código de biblioteca y la mayoría de aplicaciones con subprocesos. Tampoco tiene ningún efecto sobre la salida de los subprocesos. Sin embargo, sigue siendo un enfoque útil para muchos scripts de utilidades.

Este administrador de contexto es reentrante.

Aquí hay una solución asincrónica que usa canalizaciones de archivos.

import threading
import sys
import os

class Capturing():
    def __init__(self):
        self._stdout = None
        self._stderr = None
        self._r = None
        self._w = None
        self._thread = None
        self._on_readline_cb = None

    def _handler(self):
        while not self._w.closed:
            try:
                while True:
                    line = self._r.readline()
                    if len(line) == 0: break
                    if self._on_readline_cb: self._on_readline_cb(line)
            except:
                break

    def print(self, s, end=""):
        print(s, file=self._stdout, end=end)

    def on_readline(self, callback):
        self._on_readline_cb = callback

    def start(self):
        self._stdout = sys.stdout
        self._stderr = sys.stderr
        r, w = os.pipe()
        r, w = os.fdopen(r, 'r'), os.fdopen(w, 'w', 1)
        self._r = r
        self._w = w
        sys.stdout = self._w
        sys.stderr = self._w
        self._thread = threading.Thread(target=self._handler)
        self._thread.start()

    def stop(self):
        self._w.close()
        if self._thread: self._thread.join()
        self._r.close()
        sys.stdout = self._stdout
        sys.stderr = self._stderr

Uso de ejemplo:

from Capturing import *
import time

capturing = Capturing()

def on_read(line):
    # do something with the line
    capturing.print("got line: "+line)

capturing.on_readline(on_read)
capturing.start()
print("hello 1")
time.sleep(1)
print("hello 2")
time.sleep(1)
print("hello 3")
capturing.stop()