用io.TextIOWrapper包装一个开放的流

Question

如何打开一个开放的二进制流 - 一个Python 2 file,一个Python 3 io.BufferedReader,io.BytesIO一个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'>

但当然它依赖于从文件句柄重新打开一个真实的文件.因此,当test double是一个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.TextIOWrapperAPI.具体来说,它不继承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已经打开的字节流？

Answer 1

使用codecs.getreader生成包装器对象:

text_stream = codecs.getreader("utf-8")(bytes_stream)

适用于Python 2和Python 3.

Answer 2

根据各个论坛中的多个建议，并尝试使用标准库来满足标准，我目前的结论是，我们目前拥有的库和类型无法做到这一点。

Answer 3

事实证明，您只需要包装io.BytesIO在io.BufferedReaderPython 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，但是无论如何，管道的读取端都必须包装在Python 2的io.BufferedReader中，因此此解决方案更简单，避免了分配管道。