Оберните открытый поток с помощью 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
вокруг уже открытого потока байтов?
Ответы
Ответ 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, но работает как в обоих случаях.