如何包装一个开放的二进制流——一个 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

一个明显的 react 是在代码中有一个分支来测试流是否真的是 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。

关于python - 使用 io.TextIOWrapper 包装打开的流，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/34447623/