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