用 io.TextIOWrapper 包装一个开放的流
Wrap an open stream with io.TextIOWrapper
我如何包装一个开放的二进制流 – a Python 2 file, a Python 3 io.BufferedReader, an io.BytesIO – in an io.TextIOWrapper?
我正在尝试编写无需更改即可工作的代码：
- 运行 在 Python 2.
- 运行 在 Python 3.
- 使用标准库生成的二进制流（即我无法控制它们是什么类型）
- 使用二进制流作为测试替身（即没有文件句柄，无法重新打开）。
- 正在生成包装指定流的
io.TextIOWrapper。
io.TextIOWrapper 是必需的，因为它的 API 是标准库的其他部分所期望的。存在其他类文件类型，但不提供正确的 API.
例子
包装呈现为 subprocess.Popen.stdout 属性的二进制流：
import subprocess
import io
gnupg_subprocess = subprocess.Popen(
["gpg", "--version"], stdout=subprocess.PIPE)
gnupg_stdout = io.TextIOWrapper(gnupg_subprocess.stdout, encoding="utf-8")
在单元测试中，流被替换为 io.BytesIO 实例以控制其内容，而无需触及任何子进程或文件系统。
gnupg_subprocess.stdout = io.BytesIO("Lorem ipsum".encode("utf-8"))
这适用于 Python 3 标准库创建的流。但是，相同的代码在 Python 2:
生成的流上失败
[Python 2]
>>> type(gnupg_subprocess.stdout)
<type 'file'>
>>> gnupg_stdout = io.TextIOWrapper(gnupg_subprocess.stdout, encoding="utf-8")
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
AttributeError: 'file' object has no attribute 'readable'
不是解决方案：file
的特殊处理
一个明显的反应是在代码中有一个分支来测试流是否实际上是一个 Python 2 file 对象，并以不同于 io.* 对象的方式处理它。
对于经过良好测试的代码，这不是一个选项，因为它会创建一个分支进行单元测试 – 为了尽快 运行，不得创建任何 真实 文件系统对象 – 无法执行。
单元测试将提供测试替身，而不是真正的 file 对象。所以创建一个不会被那些测试替身执行的分支就是在打败测试套件。
不是解决方案：io.open
一些受访者建议重新打开（例如使用 io.open）底层文件句柄：
gnupg_stdout = io.open(
gnupg_subprocess.stdout.fileno(), mode='r', encoding="utf-8")
这适用于 Python 3 和 Python 2:
[Python 3]
>>> type(gnupg_subprocess.stdout)
<class '_io.BufferedReader'>
>>> gnupg_stdout = io.open(gnupg_subprocess.stdout.fileno(), mode='r', encoding="utf-8")
>>> type(gnupg_stdout)
<class '_io.TextIOWrapper'>
[Python 2]
>>> type(gnupg_subprocess.stdout)
<type 'file'>
>>> gnupg_stdout = io.open(gnupg_subprocess.stdout.fileno(), mode='r', encoding="utf-8")
>>> type(gnupg_stdout)
<type '_io.TextIOWrapper'>
当然，它依赖于从其文件句柄重新打开真实文件。因此，当测试替身是 io.BytesIO 实例时，它在单元测试中失败：
>>> gnupg_subprocess.stdout = io.BytesIO("Lorem ipsum".encode("utf-8"))
>>> type(gnupg_subprocess.stdout)
<type '_io.BytesIO'>
>>> gnupg_stdout = io.open(gnupg_subprocess.stdout.fileno(), mode='r', encoding="utf-8")
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
io.UnsupportedOperation: fileno
不是解决方案：codecs.getreader
标准库还有 codecs 模块，它提供包装器功能：
import codecs
gnupg_stdout = codecs.getreader("utf-8")(gnupg_subprocess.stdout)
很好，因为它不会尝试重新打开流。但是它没有提供 io.TextIOWrapper API。具体来说，它不继承io.IOBase并且没有encoding属性:
>>> type(gnupg_subprocess.stdout)
<type 'file'>
>>> gnupg_stdout = codecs.getreader("utf-8")(gnupg_subprocess.stdout)
>>> type(gnupg_stdout)
<type 'instance'>
>>> isinstance(gnupg_stdout, io.IOBase)
False
>>> gnupg_stdout.encoding
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/usr/lib/python2.7/codecs.py", line 643, in __getattr__
return getattr(self.stream, name)
AttributeError: '_io.BytesIO' object has no attribute 'encoding'
所以 codecs 不提供替代 io.TextIOWrapper 的对象。
怎么办？
那么我如何编写适用于 Python 2 和 Python 3 的代码，同时包含测试替身和真实对象， 包装了一个 io.TextIOWrapper围绕已经打开的字节流?
使用codecs.getreader生成包装器对象：
text_stream = codecs.getreader("utf-8")(bytes_stream)
适用于 Python 2 和 Python 3。
基于各种论坛中的多项建议，并尝试使用标准库来满足标准，我目前的结论是这无法完成 库和类型如下我们目前有。
事实证明，您只需要将 io.BytesIO 包装在 io.BufferedReader 中，它存在于 Python 2 和 Python 3 上。
import io
reader = io.BufferedReader(io.BytesIO("Lorem ipsum".encode("utf-8")))
wrapper = io.TextIOWrapper(reader)
wrapper.read() # returns Lorem ipsum
这个答案最初建议使用 os.pipe，但是管道的 read-side 必须在 Python 2 上包裹在 io.BufferedReader 中才能工作，所以这个解决方案更简单，避免分配管道。
好的，对于问题中提到的所有情况，这似乎是一个完整的解决方案，已使用 Python 2.7 和 Python 3.5 进行测试。一般的解决方案最终是 re-opening 文件描述符，但是您需要为测试替身使用管道而不是 io.BytesIO，这样您就有一个文件描述符。
import io
import subprocess
import os
# Example function, re-opens a file descriptor for UTF-8 decoding,
# reads until EOF and prints what is read.
def read_as_utf8(fileno):
fp = io.open(fileno, mode="r", encoding="utf-8", closefd=False)
print(fp.read())
fp.close()
# Subprocess
gpg = subprocess.Popen(["gpg", "--version"], stdout=subprocess.PIPE)
read_as_utf8(gpg.stdout.fileno())
# Normal file (contains "Lorem ipsum." as UTF-8 bytes)
normal_file = open("loremipsum.txt", "rb")
read_as_utf8(normal_file.fileno()) # prints "Lorem ipsum."
# Pipe (for test harness - write whatever you want into the pipe)
pipe_r, pipe_w = os.pipe()
os.write(pipe_w, "Lorem ipsum.".encode("utf-8"))
os.close(pipe_w)
read_as_utf8(pipe_r) # prints "Lorem ipsum."
os.close(pipe_r)
我也需要这个，但是根据此处的线程，我确定仅使用 Python 2 的 io 模块是不可能的。虽然这违反了您的 "Special treatment for file" 规则，但我采用的技术是为 file（下面的代码）创建一个非常薄的包装器，然后可以将其包装在 io.BufferedReader 中，后者又可以传递给 io.TextIOWrapper 构造函数。单元测试会很痛苦，因为显然无法在 Python 3.
上测试新代码路径
顺便说一句，open()的结果可以直接传递给Python3中的io.TextIOWrapper是因为binary-modeopen()实际上returns 一个 io.BufferedReader 实例开始（至少在 Python 3.4 上，这是我当时正在测试的地方）。
import io
import six # for six.PY2
if six.PY2:
class _ReadableWrapper(object):
def __init__(self, raw):
self._raw = raw
def readable(self):
return True
def writable(self):
return False
def seekable(self):
return True
def __getattr__(self, name):
return getattr(self._raw, name)
def wrap_text(stream, *args, **kwargs):
# Note: order important here, as 'file' doesn't exist in Python 3
if six.PY2 and isinstance(stream, file):
stream = io.BufferedReader(_ReadableWrapper(stream))
return io.TextIOWrapper(stream)
至少这很小，所以希望它能最大限度地减少无法轻易进行单元测试的部分的暴露。
这是我在 python 2.7 和 python 3.6 中测试过的一些代码。
这里的关键是您需要先在之前的流中使用 detach()。这不会关闭底层文件，它只是撕掉原始流对象以便可以重用。 detach() 将 return 一个可以用 TextIOWrapper 包装的对象。
举个例子，我以二进制读取模式打开一个文件，像这样读取它，然后我通过 io.TextIOWrapper.
切换到 UTF-8 解码文本流
我把这个例子保存为这个-file.py
import io
fileName = 'this-file.py'
fp = io.open(fileName,'rb')
fp.seek(20)
someBytes = fp.read(10)
print(type(someBytes) + len(someBytes))
# now let's do some wrapping to get a new text (non-binary) stream
pos = fp.tell() # we're about to lose our position, so let's save it
newStream = io.TextIOWrapper(fp.detach(),'utf-8') # FYI -- fp is now unusable
newStream.seek(pos)
theRest = newStream.read()
print(type(theRest), len(theRest))
这是我 运行 同时使用 python2 和 python3 时得到的结果。
$ python2.7 this-file.py
(<type 'str'>, 10)
(<type 'unicode'>, 406)
$ python3.6 this-file.py
<class 'bytes'> 10
<class 'str'> 406
显然打印语法不同，正如预期的那样，python 版本之间的变量类型不同，但在这两种情况下都可以正常工作。
我如何包装一个开放的二进制流 – a Python 2 file, a Python 3 io.BufferedReader, an io.BytesIO – in an io.TextIOWrapper?
我正在尝试编写无需更改即可工作的代码：
- 运行 在 Python 2.
- 运行 在 Python 3.
- 使用标准库生成的二进制流（即我无法控制它们是什么类型）
- 使用二进制流作为测试替身（即没有文件句柄，无法重新打开）。
- 正在生成包装指定流的
io.TextIOWrapper。
io.TextIOWrapper 是必需的，因为它的 API 是标准库的其他部分所期望的。存在其他类文件类型，但不提供正确的 API.
例子
包装呈现为 subprocess.Popen.stdout 属性的二进制流：
import subprocess
import io
gnupg_subprocess = subprocess.Popen(
["gpg", "--version"], stdout=subprocess.PIPE)
gnupg_stdout = io.TextIOWrapper(gnupg_subprocess.stdout, encoding="utf-8")
在单元测试中，流被替换为 io.BytesIO 实例以控制其内容，而无需触及任何子进程或文件系统。
gnupg_subprocess.stdout = io.BytesIO("Lorem ipsum".encode("utf-8"))
这适用于 Python 3 标准库创建的流。但是，相同的代码在 Python 2:
生成的流上失败[Python 2]
>>> type(gnupg_subprocess.stdout)
<type 'file'>
>>> gnupg_stdout = io.TextIOWrapper(gnupg_subprocess.stdout, encoding="utf-8")
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
AttributeError: 'file' object has no attribute 'readable'
不是解决方案：file
的特殊处理
一个明显的反应是在代码中有一个分支来测试流是否实际上是一个 Python 2 file 对象，并以不同于 io.* 对象的方式处理它。
对于经过良好测试的代码，这不是一个选项，因为它会创建一个分支进行单元测试 – 为了尽快 运行，不得创建任何 真实 文件系统对象 – 无法执行。
单元测试将提供测试替身，而不是真正的 file 对象。所以创建一个不会被那些测试替身执行的分支就是在打败测试套件。
不是解决方案：io.open
一些受访者建议重新打开（例如使用 io.open）底层文件句柄：
gnupg_stdout = io.open(
gnupg_subprocess.stdout.fileno(), mode='r', encoding="utf-8")
这适用于 Python 3 和 Python 2:
[Python 3]
>>> type(gnupg_subprocess.stdout)
<class '_io.BufferedReader'>
>>> gnupg_stdout = io.open(gnupg_subprocess.stdout.fileno(), mode='r', encoding="utf-8")
>>> type(gnupg_stdout)
<class '_io.TextIOWrapper'>
[Python 2]
>>> type(gnupg_subprocess.stdout)
<type 'file'>
>>> gnupg_stdout = io.open(gnupg_subprocess.stdout.fileno(), mode='r', encoding="utf-8")
>>> type(gnupg_stdout)
<type '_io.TextIOWrapper'>
当然，它依赖于从其文件句柄重新打开真实文件。因此，当测试替身是 io.BytesIO 实例时，它在单元测试中失败：
>>> gnupg_subprocess.stdout = io.BytesIO("Lorem ipsum".encode("utf-8"))
>>> type(gnupg_subprocess.stdout)
<type '_io.BytesIO'>
>>> gnupg_stdout = io.open(gnupg_subprocess.stdout.fileno(), mode='r', encoding="utf-8")
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
io.UnsupportedOperation: fileno
不是解决方案：codecs.getreader
标准库还有 codecs 模块，它提供包装器功能：
import codecs
gnupg_stdout = codecs.getreader("utf-8")(gnupg_subprocess.stdout)
很好，因为它不会尝试重新打开流。但是它没有提供 io.TextIOWrapper API。具体来说，它不继承io.IOBase并且没有encoding属性:
>>> type(gnupg_subprocess.stdout)
<type 'file'>
>>> gnupg_stdout = codecs.getreader("utf-8")(gnupg_subprocess.stdout)
>>> type(gnupg_stdout)
<type 'instance'>
>>> isinstance(gnupg_stdout, io.IOBase)
False
>>> gnupg_stdout.encoding
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/usr/lib/python2.7/codecs.py", line 643, in __getattr__
return getattr(self.stream, name)
AttributeError: '_io.BytesIO' object has no attribute 'encoding'
所以 codecs 不提供替代 io.TextIOWrapper 的对象。
怎么办？
那么我如何编写适用于 Python 2 和 Python 3 的代码，同时包含测试替身和真实对象， 包装了一个 io.TextIOWrapper围绕已经打开的字节流?
使用codecs.getreader生成包装器对象：
text_stream = codecs.getreader("utf-8")(bytes_stream)
适用于 Python 2 和 Python 3。
基于各种论坛中的多项建议，并尝试使用标准库来满足标准，我目前的结论是这无法完成 库和类型如下我们目前有。
事实证明，您只需要将 io.BytesIO 包装在 io.BufferedReader 中，它存在于 Python 2 和 Python 3 上。
import io
reader = io.BufferedReader(io.BytesIO("Lorem ipsum".encode("utf-8")))
wrapper = io.TextIOWrapper(reader)
wrapper.read() # returns Lorem ipsum
这个答案最初建议使用 os.pipe，但是管道的 read-side 必须在 Python 2 上包裹在 io.BufferedReader 中才能工作，所以这个解决方案更简单，避免分配管道。
好的，对于问题中提到的所有情况，这似乎是一个完整的解决方案，已使用 Python 2.7 和 Python 3.5 进行测试。一般的解决方案最终是 re-opening 文件描述符，但是您需要为测试替身使用管道而不是 io.BytesIO，这样您就有一个文件描述符。
import io
import subprocess
import os
# Example function, re-opens a file descriptor for UTF-8 decoding,
# reads until EOF and prints what is read.
def read_as_utf8(fileno):
fp = io.open(fileno, mode="r", encoding="utf-8", closefd=False)
print(fp.read())
fp.close()
# Subprocess
gpg = subprocess.Popen(["gpg", "--version"], stdout=subprocess.PIPE)
read_as_utf8(gpg.stdout.fileno())
# Normal file (contains "Lorem ipsum." as UTF-8 bytes)
normal_file = open("loremipsum.txt", "rb")
read_as_utf8(normal_file.fileno()) # prints "Lorem ipsum."
# Pipe (for test harness - write whatever you want into the pipe)
pipe_r, pipe_w = os.pipe()
os.write(pipe_w, "Lorem ipsum.".encode("utf-8"))
os.close(pipe_w)
read_as_utf8(pipe_r) # prints "Lorem ipsum."
os.close(pipe_r)
我也需要这个，但是根据此处的线程，我确定仅使用 Python 2 的 io 模块是不可能的。虽然这违反了您的 "Special treatment for file" 规则，但我采用的技术是为 file（下面的代码）创建一个非常薄的包装器，然后可以将其包装在 io.BufferedReader 中，后者又可以传递给 io.TextIOWrapper 构造函数。单元测试会很痛苦，因为显然无法在 Python 3.
顺便说一句，open()的结果可以直接传递给Python3中的io.TextIOWrapper是因为binary-modeopen()实际上returns 一个 io.BufferedReader 实例开始（至少在 Python 3.4 上，这是我当时正在测试的地方）。
import io
import six # for six.PY2
if six.PY2:
class _ReadableWrapper(object):
def __init__(self, raw):
self._raw = raw
def readable(self):
return True
def writable(self):
return False
def seekable(self):
return True
def __getattr__(self, name):
return getattr(self._raw, name)
def wrap_text(stream, *args, **kwargs):
# Note: order important here, as 'file' doesn't exist in Python 3
if six.PY2 and isinstance(stream, file):
stream = io.BufferedReader(_ReadableWrapper(stream))
return io.TextIOWrapper(stream)
至少这很小，所以希望它能最大限度地减少无法轻易进行单元测试的部分的暴露。
这是我在 python 2.7 和 python 3.6 中测试过的一些代码。
这里的关键是您需要先在之前的流中使用 detach()。这不会关闭底层文件，它只是撕掉原始流对象以便可以重用。 detach() 将 return 一个可以用 TextIOWrapper 包装的对象。
举个例子，我以二进制读取模式打开一个文件，像这样读取它，然后我通过 io.TextIOWrapper.
切换到 UTF-8 解码文本流我把这个例子保存为这个-file.py
import io
fileName = 'this-file.py'
fp = io.open(fileName,'rb')
fp.seek(20)
someBytes = fp.read(10)
print(type(someBytes) + len(someBytes))
# now let's do some wrapping to get a new text (non-binary) stream
pos = fp.tell() # we're about to lose our position, so let's save it
newStream = io.TextIOWrapper(fp.detach(),'utf-8') # FYI -- fp is now unusable
newStream.seek(pos)
theRest = newStream.read()
print(type(theRest), len(theRest))
这是我 运行 同时使用 python2 和 python3 时得到的结果。
$ python2.7 this-file.py
(<type 'str'>, 10)
(<type 'unicode'>, 406)
$ python3.6 this-file.py
<class 'bytes'> 10
<class 'str'> 406
显然打印语法不同，正如预期的那样，python 版本之间的变量类型不同，但在这两种情况下都可以正常工作。