当前位置:  开发笔记 > 编程语言 > 正文

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

如何解决《用io.TextIOWrapper包装一个开放的流》经验,为你挑选了2个好方法。

如何打开一个开放的二进制流 - 一个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)

>>> gnupg_stdout = io.TextIOWrapper(gnupg_subprocess.stdout, encoding="utf-8")
Traceback (most recent call last):
  File "", line 1, in 
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)

>>> gnupg_stdout = io.open(gnupg_subprocess.stdout.fileno(), mode='r', encoding="utf-8")
>>> type(gnupg_stdout)

[Python 2]
>>> type(gnupg_subprocess.stdout)

>>> gnupg_stdout = io.open(gnupg_subprocess.stdout.fileno(), mode='r', encoding="utf-8")
>>> type(gnupg_stdout)

但当然它依赖于从文件句柄重新打开一个真实的文件.因此,当test double是一个io.BytesIO实例时,它在单元测试中失败:

>>> gnupg_subprocess.stdout = io.BytesIO("Lorem ipsum".encode("utf-8"))
>>> type(gnupg_subprocess.stdout)

>>> gnupg_stdout = io.open(gnupg_subprocess.stdout.fileno(), mode='r', encoding="utf-8")
Traceback (most recent call last):
  File "", line 1, in 
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)

>>> gnupg_stdout = codecs.getreader("utf-8")(gnupg_subprocess.stdout)
>>> type(gnupg_stdout)

>>> isinstance(gnupg_stdout, io.IOBase)
False
>>> gnupg_stdout.encoding
Traceback (most recent call last):
  File "", line 1, in 
  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已经打开的字节流



1> jbg..:

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

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

适用于Python 2和Python 3.


对我来说就像一个魅力.使用此技术与`csv`包和`boto`一起从S3流式传输CSV文件.

2> jbg..:

事实证明,您只需要包装io.BytesIOio.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中,因此此解决方案更简单,避免了分配管道。

推荐阅读
mobiledu2402852413
这个屌丝很懒,什么也没留下!
DevBox开发工具箱 | 专业的在线开发工具网站    京公网安备 11010802040832号  |  京ICP备19059560号-6
Copyright © 1998 - 2020 DevBox.CN. All Rights Reserved devBox.cn 开发工具箱 版权所有