我可以将标准输出重定向到某种字符串缓冲区吗？

我使用 python 的 ftplib来编写一个小型 FTP 客户端，但是包中的一些函数不返回字符串输出，而是打印到 stdout。我想将 stdout重定向到一个对象，这样我就能够从中读取输出。

我知道 stdout可以重定向到任何常规文件:

stdout = open("file", "a")

但我更喜欢不使用本地驱动器的方法。

我正在寻找类似于 Java 中的 BufferedReader的东西，它可以用来将缓冲区封装到流中。

164417 次浏览
小开

使用 pipe()并写入适当的文件描述符。

Https://docs.python.org/library/os.html#file-descriptor-operations

小开
最佳答案 
from cStringIO import StringIO # Python3 use: from io import StringIO
import sys


old_stdout = sys.stdout
sys.stdout = mystdout = StringIO()


# blah blah lots of code ...


sys.stdout = old_stdout


# examine mystdout.getvalue()
小开

只是为了补充 Ned 上面的答案: 您可以使用它将输出重定向到 实现 write (str)方法的任何对象。

这可以用来在 GUI 应用程序中很好地“捕获”stdout 输出。

下面是 PyQt 中一个愚蠢的例子:

import sys
from PyQt4 import QtGui


class OutputWindow(QtGui.QPlainTextEdit):
def write(self, txt):
self.appendPlainText(str(txt))


app = QtGui.QApplication(sys.argv)
out = OutputWindow()
sys.stdout=out
out.show()
print "hello world !"
小开

从 Python 2.6开始，您可以使用 IO 模块中实现 TextIOBaseAPI的任何东西作为替代。 此解决方案还允许您在 Python 3中使用 sys.stdout.buffer.write()将(已经)编码的字节字符串写入标准输出(请参见 Python 3中的 stdout)。 那么使用 StringIO将不起作用，因为 sys.stdout.encoding和 sys.stdout.buffer都不可用。

使用 TextIOWrapper 的解决方案:

import sys
from io import TextIOWrapper, BytesIO


# setup the environment
old_stdout = sys.stdout
sys.stdout = TextIOWrapper(BytesIO(), sys.stdout.encoding)


# do something that writes to stdout or stdout.buffer


# get output
sys.stdout.seek(0)      # jump to the start
out = sys.stdout.read() # read output


# restore stdout
sys.stdout.close()
sys.stdout = old_stdout

该解决方案适用于 Python 2 > = 2.6和 Python 3。

请注意，我们的新 sys.stdout.write()只接受 unicode 字符串，而 sys.stdout.buffer.write()只接受字节字符串。 旧代码可能不会出现这种情况，但是对于构建在 Python 2和3上运行而不进行更改的代码通常会出现这种情况，这种情况通常也会使用 sys.stdout.buffer。

您可以构建一个接受 Unicode 和字节字符串作为 write()的轻微变体:

class StdoutBuffer(TextIOWrapper):
def write(self, string):
try:
return super(StdoutBuffer, self).write(string)
except TypeError:
# redirect encoded byte strings directly to buffer
return super(StdoutBuffer, self).buffer.write(string)

您不必将缓冲区的编码设置为 sys.stdout.coding，但是在使用这种方法测试/比较脚本输出时，这会有所帮助。

小开

Python 3.4 + 中有一个 contextlib.redirect_stdout()功能:

import io
from contextlib import redirect_stdout


with io.StringIO() as buf, redirect_stdout(buf):
print('redirected')
output = buf.getvalue()

这是 演示如何在旧的 Python 版本上实现它的代码示例。

小开

即使有异常，这个方法也会恢复 sys.stdout。

import io
import sys


real_stdout = sys.stdout
fake_stdout = io.BytesIO()   # or perhaps io.StringIO()
try:
sys.stdout = fake_stdout
# do what you have to do to create some output
finally:
sys.stdout = real_stdout
output_string = fake_stdout.getvalue()
fake_stdout.close()
# do what you want with the output_string

在 Python 2.7.10中使用 io.BytesIO()进行测试

使用 io.StringIO()在 Python 3.6.4中进行测试

Bob，如果您觉得修改/扩展代码实验中的任何东西可能会在任何意义上变得有趣，那么可以添加它，否则 < strong > 可以随意删除它

广告信息备忘录... 一些意见从扩展实验期间寻找一些可行的机制“抓取”输出，由 numexpr.print_versions()直接到 <stdout>(在需要清洁 GUI 和收集细节进行调试报告)

# THIS WORKS AS HELL: as Bob Stein proposed years ago:
#  py2 SURPRISEDaBIT:
#
import io
import sys
#
real_stdout = sys.stdout                        #           PUSH <stdout> ( store to REAL_ )
fake_stdout = io.BytesIO()                      #           .DEF FAKE_
try:                                            # FUSED .TRY:
sys.stdout.flush()                          #           .flush() before
sys.stdout = fake_stdout                    #           .SET <stdout> to use FAKE_
# ----------------------------------------- #           +    do what you gotta do to create some output
print 123456789                             #           +
import  numexpr                             #           +
QuantFX.numexpr.__version__                 #           + [3] via fake_stdout re-assignment, as was bufferred + "late" deferred .get_value()-read into print, to finally reach -> real_stdout
QuantFX.numexpr.print_versions()            #           + [4] via fake_stdout re-assignment, as was bufferred + "late" deferred .get_value()-read into print, to finally reach -> real_stdout
_ = os.system( 'echo os.system() redir-ed' )#           + [1] via real_stdout                                 + "late" deferred .get_value()-read into print, to finally reach -> real_stdout, if not ( _ = )-caught from RET-d "byteswritten" / avoided from being injected int fake_stdout
_ = os.write(  sys.stderr.fileno(),         #           + [2] via      stderr                                 + "late" deferred .get_value()-read into print, to finally reach -> real_stdout, if not ( _ = )-caught from RET-d "byteswritten" / avoided from being injected int fake_stdout
b'os.write()  redir-ed' )#  *OTHERWISE, if via fake_stdout, EXC <_io.BytesIO object at 0x02C0BB10> Traceback (most recent call last):
# ----------------------------------------- #           ?                              io.UnsupportedOperation: fileno
#'''                                                    ? YET:        <_io.BytesIO object at 0x02C0BB10> has a .fileno() method listed
#>>> 'fileno' in dir( sys.stdout )       -> True        ? HAS IT ADVERTISED,
#>>> pass;            sys.stdout.fileno  -> <built-in method fileno of _io.BytesIO object at 0x02C0BB10>
#>>> pass;            sys.stdout.fileno()-> Traceback (most recent call last):
#                                             File "<stdin>", line 1, in <module>
#                                           io.UnsupportedOperation: fileno
#                                                       ? BUT REFUSES TO USE IT
#'''
finally:                                        # == FINALLY:
sys.stdout.flush()                          #           .flush() before ret'd back REAL_
sys.stdout = real_stdout                    #           .SET <stdout> to use POP'd REAL_
sys.stdout.flush()                          #           .flush() after  ret'd back REAL_
out_string = fake_stdout.getvalue()         #           .GET string           from FAKE_
fake_stdout.close()                         #                <FD>.close()
# +++++++++++++++++++++++++++++++++++++     # do what you want with the out_string
#
print "\n{0:}\n{1:}{0:}".format( 60 * "/\\",# "LATE" deferred print the out_string at the very end reached -> real_stdout
out_string #
)
'''
PASS'd:::::
...
os.system() redir-ed
os.write()  redir-ed
/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\
123456789
'2.5'
-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
Numexpr version:   2.5
NumPy version:     1.10.4
Python version:    2.7.13 |Anaconda 4.0.0 (32-bit)| (default, May 11 2017, 14:07:41) [MSC v.1500 32 bit (Intel)]
AMD/Intel CPU?     True
VML available?     True
VML/MKL version:   Intel(R) Math Kernel Library Version 11.3.1 Product Build 20151021 for 32-bit applications
Number of threads used by default: 4 (out of 4 detected cores)
-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\
>>>


EXC'd :::::
...
os.system() redir-ed
/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\
123456789
'2.5'
-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
Numexpr version:   2.5
NumPy version:     1.10.4
Python version:    2.7.13 |Anaconda 4.0.0 (32-bit)| (default, May 11 2017, 14:07:41) [MSC v.1500 32 bit (Intel)]
AMD/Intel CPU?     True
VML available?     True
VML/MKL version:   Intel(R) Math Kernel Library Version 11.3.1 Product Build 20151021 for 32-bit applications
Number of threads used by default: 4 (out of 4 detected cores)
-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\


Traceback (most recent call last):
File "<stdin>", line 9, in <module>
io.UnsupportedOperation: fileno
'''
小开

在 Python 3.6中，没有了 StringIO和 cStringIO模块，应该使用 io.StringIO:

import sys
from io import StringIO


old_stdout = sys.stdout
old_stderr = sys.stderr
my_stdout = sys.stdout = StringIO()
my_stderr = sys.stderr = StringIO()


# blah blah lots of code ...


sys.stdout = self.old_stdout
sys.stderr = self.old_stderr


// if you want to see the value of redirect output, be sure the std output is turn back
print(my_stdout.getvalue())
print(my_stderr.getvalue())


my_stdout.close()
my_stderr.close()
小开

Python3的上下文管理器:

import sys
from io import StringIO




class RedirectedStdout:
def __init__(self):
self._stdout = None
self._string_io = None


def __enter__(self):
self._stdout = sys.stdout
sys.stdout = self._string_io = StringIO()
return self


def __exit__(self, type, value, traceback):
sys.stdout = self._stdout


def __str__(self):
return self._string_io.getvalue()

使用方法如下:

>>> with RedirectedStdout() as out:
>>>     print('asdf')
>>>     s = str(out)
>>>     print('bsdf')
>>> print(s, out)
'asdf\n' 'asdf\nbsdf\n'
小开

这里有另一个想法。contextlib.redirect_stdout和 io.StringIO()就像 记录在案一样很棒，但是对于每天的使用来说还是有点冗长。下面是如何通过子类化 contextlib.redirect_stdout使其成为一行程序:

import sys
import io
from contextlib import redirect_stdout


class capture(redirect_stdout):


def __init__(self):
self.f = io.StringIO()
self._new_target = self.f
self._old_targets = []  # verbatim from parent class


def __enter__(self):
self._old_targets.append(getattr(sys, self._stream))  # verbatim from parent class
setattr(sys, self._stream, self._new_target)  # verbatim from parent class
return self  # instead of self._new_target in the parent class


def __repr__(self):
return self.f.getvalue()

因为 _ _ enter _ _ 返回 self，所以在 with 块退出之后可以使用上下文管理器对象。此外，由于使用了 _ _ repr _ _ 方法，上下文管理器对象的字符串表示实际上是 stdout。所以现在你有了,

with capture() as message:
print('Hello World!')
print(str(message)=='Hello World!\n')  # returns True

提交答案