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

Question

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

Как я могу обернуть открытый двоичный поток - Python 2 file Питон 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`

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

55

python python-3.x unit-testing character-encoding python-2.x

Источник

user70157 24 дек '15 в 05:17

6 ответов

Решение

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

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

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

30

Источник

user1174521 29 дек '15 в 13:11

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

16

Источник

user1174521 30 дек '15 в 08:38

Хорошо, похоже, это полное решение для всех случаев, упомянутых в вопросе, протестированных с 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

Источник

user1174521 30 дек '15 в 13:59

Мне это тоже нужно, но, основываясь на этой теме, я решил, что это невозможно, используя только Python 2 io модуль. Пока это ломает твой file"Правило, техника, которую я использовал, заключалась в создании очень тонкой обертки для file (код ниже), который затем может быть обернут в io.BufferedReaderкоторый в свою очередь может быть передан io.TextIOWrapper конструктор. Модульное тестирование будет болезненным, так как новый путь кода не может быть протестирован на 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)

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

2

Источник

user7589408 19 фев '17 в 18:13

Вот некоторый код, который я протестировал в 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'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, но в обоих случаях работает так, как должно.

1

Источник

user5433297 01 мар '17 в 23:15

Другие вопросы по тегам python python-3.x unit-testing character-encoding python-2.x

user70157 30 дек '15 в 07:47 2015-12-30 07:47 · Accepted Answer · 2015-12-30 07:47

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

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

пример

Не решение: специальное лечение для file

Не решение io.open

Не решение codecs.getreader

Что делать?

6 ответов

Не решение: специальное лечение для `file`

Не решение `io.open`

Не решение `codecs.getreader`