Подтвердить что ты не робот

Оберните открытый поток с помощью io.TextIOWrapper

Как я могу обернуть открытый двоичный поток - Python 2 file, Python 3 io.BufferedReader, a 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'>

Но, конечно, полагается на повторное открытие реального файла из дескриптора файла. Таким образом, он не работает в модульных тестах, когда тестовый 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)

Это хорошо, потому что он не пытается повторно открыть поток. Но он не предоставляет API io.TextIOWrapper. В частности, он не наследует 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 вокруг уже открытого потока байтов?

4b9b3361

Ответ 1

Основываясь на нескольких предложениях на разных форумах и экспериментируя со стандартной библиотекой, чтобы соответствовать критериям, мой текущий вывод не может быть выполнен с помощью библиотеки и типов, которые у нас есть в настоящее время.

Ответ 2

Используйте codecs.getreader для создания объекта-оболочки:

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

Работает на Python 2 и Python 3.

Ответ 3

Хорошо, это, кажется, полное решение для всех случаев, упомянутых в вопросе, проверенных с помощью Python 2.7 и Python 3.5. Общее решение закончило тем, что повторно открыло дескриптор файла, но вместо 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)

Ответ 4

Оказывается, вам просто нужно обернуть 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, но чтение стороны трубы должно было быть завернуто в io.BufferedReader на Python 2 в любом случае, чтобы работать, поэтому это решение проще и позволяет выделять канал.

Ответ 5

Мне тоже это нужно, но, основываясь на этом потоке, я решил, что это невозможно, используя только модуль Python 2 io. Хотя это нарушает правило "Специальное лечение для file", техника, с которой я работал, заключалась в создании чрезвычайно тонкой обертки для file (код ниже), который затем можно было бы обернуть в io.BufferedReader, что, в свою очередь, может быть передается конструктору io.TextIOWrapper. Это будет болью до unit test, так как, очевидно, новый код не может быть протестирован на Python 3.

Кстати, причина, по которой результаты open() могут быть переданы непосредственно в io.TextIOWrapper в Python 3, состоит в том, что двоичный режим 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)

По крайней мере, это мало, поэтому, надеюсь, он минимизирует экспозицию для деталей, которые не могут быть легко протестированы.

Ответ 6

Вот некоторый код, который я тестировал как в python 2.7, так и в python 3.6.

Ключевым моментом здесь является то, что сначала необходимо использовать detach() в своем предыдущем потоке. Это не закрывает основной файл, он просто вырывает необработанный объект потока, чтобы его можно было повторно использовать. Функция detach() вернет объект, который обертывается с помощью TextIOWrapper.

В качестве примера здесь я открываю файл в режиме двоичного чтения, читаю его так, а затем переключаюсь на декодированный текстовый поток UTF-8 через io.TextIOWrapper.

Я сохранил этот пример как this-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 do some wrapping to get a new text (non-binary) stream
pos = fp.tell() # we're about to lose our position, so let 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, но работает как в обоих случаях.