Сообщить об ошибке.

Паттерны использования io.TextIOWrapper() в Python

Содержание:

io.TextIOWrapper - это текстовый слой (строки str) поверх бинарного потока (BufferedIOBase / RawIOBase). Все open(..., encoding=...) под капотом - именно TextIOWrapper.

Ниже - самые полезные паттерны для io.TextIOWrapper.

Явная сборка стека: FileIO => Buffered => TextIOWrapper

Обычно ты просто пишешь:

with open("log.txt", "w", encoding="utf-8") as f:
    f.write("привет\n")

Но эквивалентный "ручной" стек:

import io

raw = io.FileIO("log.txt", "w")           # RawIOBase
buf = io.BufferedWriter(raw)             # BufferedIOBase
text = io.TextIOWrapper(
    buf,
    encoding="utf-8",
    newline="\n",         # как нормализовать переводы строки
    line_buffering=True,  # flush при '\n'
)

text.write("привет\n")
text.flush()
text.close()

Паттерн: когда нужен контроль всех слоёв (логирование, шифрование, нестандартный буфер) - собираешь стек руками, верхний слой всегда TextIOWrapper.

Текст поверх сокета / пайпа / subprocess

Любой бинарный поток (сокет, stdout процесса) можно превратить в строковый через TextIOWrapper.

Пример: текстовый клиент поверх сокета

import socket, io

sock = socket.create_connection(("example.com", 80), timeout=5)

## бинарный file-like над сокетом
raw = sock.makefile("rwb", buffering=0)      # уже BufferedRWPair
text = io.TextIOWrapper(
    raw,
    encoding="utf-8",
    newline="\n",
    line_buffering=True,
)

text.write("GET / HTTP/1.0\r\nHost: example.com\r\n\r\n")
text.flush()

status_line = text.readline()
print("STATUS:", status_line.rstrip("\n"))
text.close()
sock.close()

Паттерн: любой socket / Popen(..., stdout=PIPE, stdin=PIPE) => оборачиваем его байтовый поток в TextIOWrapper и работаем как с обычным текстовым файлом.

Настройка newline (универсальные переводы строк)

Параметр newline управляет тем, как \r, \n, \r\n обрабатываются.

Основные режимы:

  • newline=None (по умолчанию):чтение - универсальные переводы строк: \r, \n, \r\n => \n в str;запись - \n в строке => системный os.linesep.
  • newline="\n":чтение - не трогаем байты, но \r\n => \n (тонкости реализации, но по сути нормализует);запись - \n всегда пишется как байтовый \n.
  • newline="":чтение - никакой магии, все комбинации \r, \n возвращаются как есть.

Пример: строжайший контроль над \n

import io

raw = open("data.txt", "rb")
text = io.TextIOWrapper(
    raw,
    encoding="utf-8",
    newline="",  # не трогать newlines
)

for line in text:
    # line может заканчиватьcя на '\n', '\r\n' или '\r'
    print(repr(line))

Паттерн: протоколы / форматы, где важно различать \r и \n - используем newline="".

Управление буферизацией: line_buffering и flush

Для логов и интерактивных протоколов хочется, чтобы данные улетали сразу после \n.

import io, sys

text = io.TextIOWrapper(
    sys.stdout.buffer,      # бинарный stdout
    encoding="utf-8",
    newline="\n",
    line_buffering=True,    # flush при '\n'
)

text.write("hello\n")       # почти мгновенно в терминале

Паттерн: line_buffering=True + newline="\n" для логов / REPL / чатов.

Перекодировщик (transcoding): один поток, другая кодировка

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

Пример: CP1251 => UTF-8

import io

## входной файл в cp1251
raw_in = open("input_cp1251.txt", "rb")
text_in = io.TextIOWrapper(raw_in, encoding="cp1251")

## выходной файл в utf-8
raw_out = open("output_utf8.txt", "wb")
text_out = io.TextIOWrapper(raw_out, encoding="utf-8")

for line in text_in:
    text_out.write(line)

text_out.flush()
text_in.close()
text_out.close()

Паттерн: TextIOWrapper как чистый строковый слой, а перенастройка кодировки делается сменой обёртки.

Переобёртывание sys.stdout / sys.stdin с нужной кодировкой

На некоторых системах стандартная кодировка консоли неудобна (cp1251, ascii и т.п.). Можно сделать свой stdout:

import sys, io

## sys.stdout.buffer - бинарный stdout (BufferedWriter)
sys.stdout = io.TextIOWrapper(
    sys.stdout.buffer,
    encoding="utf-8",
    newline="\n",
    line_buffering=True,
)

print("Теперь stdout в UTF-8")

Или stdin:

sys.stdin = io.TextIOWrapper(
    sys.stdin.buffer,
    encoding="utf-8",
    newline=None,
)

Паттерн: когда окружение даёт "кривую" кодировку, но ты хочешь в коде всегда работать с UTF-8.

Текстовый буфер поверх BytesIO

Иногда нужно ин-мемори текстовый "файл", но с явной кодировкой.

import io

bin_buf = io.BytesIO()
text = io.TextIOWrapper(bin_buf, encoding="utf-8")

text.write("привет\nмир")
text.flush()

## Получить байты:
data = bin_buf.getvalue()
print(data)  # b'\xd0\xbf\xd1\x80...'

## Прочитать обратно как текст:
bin_buf.seek(0)
print(text.read())  # "привет\nмир"

Паттерн: BytesIO + TextIOWrapper => полностью файловый текстовый API, но всё в памяти.

Адаптер над своим RawIOBase / BufferedIOBase

Как только есть свой RawIOBase (сокет, шифрованный поток, лимитер и т.п.), то можно получить текстовый интерфейс:

import io

class MyRaw(io.RawIOBase):
    ...  # твоя реализация readinto/write

raw = MyRaw(...)
buf = io.BufferedRWPair(
    io.BufferedReader(raw),
    io.BufferedWriter(raw),
)
text = io.TextIOWrapper(buf, encoding="utf-8")

text.write("hello\n")
text.flush()
resp = text.readline()

Паттерн: не делать вручную encode/decode в каждой операции, а сразу обернуть всё в TextIOWrapper.

Ошибки кодировки: errors="ignore"|"replace"|...

При работе с "грязным" вводом удобно смягчить поведение:

import io

raw = open("maybe-broken.txt", "rb")
text = io.TextIOWrapper(
    raw,
    encoding="utf-8",
    errors="replace",   # битые байты => �
)

for line in text:
    ...

Паттерн: при парсинге логов/сетевых протоколов, где возможен мусор, используем errors="replace" или "ignore".