如何使用io.TextIOWrapper打包流?

内容来源于 Stack Overflow,并遵循CC BY-SA 3.0许可协议进行翻译与使用

  • 回答 (6)
  • 关注 (0)
  • 查看 (2429)

我如何打包一个开放的二进制流 - 一个Python 2文件,一个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'
提问于
用户回答回答于

这里的关键是需要首先在前面的流上使用DETACH()。这不会关闭基础文件,它只是删除原始流对象,以便可以重用它。DETACH()将返回一个可以用TextIOWrapper包装的对象。

我将这个示例保存为-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))

这是我在运行python 2和python 3时得到的结果。

$ python2.7 this-file.py 
(<type 'str'>, 10)
(<type 'unicode'>, 406)
$ python3.6 this-file.py 
<class 'bytes'> 10
<class 'str'> 406
用户回答回答于

我也需要这个,但基于这里的线索,我确定使用Python 2的io模块是不可能的。 虽然这打破了对文件的“特殊待遇”规则,但我使用的技术是为文件(代码如下)创建一个非常薄的包装器,然后可以将其包装在io.BufferedReader中,然后可以将其传递给io .TextIOWrapper构造函数。 单元测试将是一件痛苦的事情,很显然新的代码路径不能在Python 3上测试。

顺便说一句,open()的结果可以直接传递给Python 3中的io.TextIOWrapper的原因是因为二进制模式open()实际上返回一个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)
用户回答回答于

原来你只需要把你的io.BytesIOio.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
用户回答回答于

需要使用管道进行双测试,以便有一个文件描述符。

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)
用户回答回答于

根据各种论坛的多种建议,并试用符合标准的标准库,我目前的结论是,这不能用我们目前拥有的库和类型来完成。

用户回答回答于

使用codecs.getreader若要生成包装器对象,请执行以下操作:

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

工作在Python 2和Python 3上。

扫码关注云+社区

领取腾讯云代金券