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

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

Содержание:

io.TextIOBase - это базовый класс для текстовых потоков (строки str, а не байты). Обычно ты работаешь с его наследниками:

  • io.TextIOWrapper (текст поверх бинарного потока),
  • sys.stdin / sys.stdout / open(..., encoding=...).

Но иногда полезно наследоваться от TextIOBase или использовать его как общий тип.

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

Общий тип для "любой текстовый поток"

Функции, которые работают с любым текстовым потоком (файлы, StringIO, TextIOWrapper, кастомные классы), удобно типизировать через TextIOBase.

from io import TextIOBase

def copy_text(src: TextIOBase, dst: TextIOBase, chunk_size: int = 4096) -> None:
    while True:
        chunk = src.read(chunk_size)
        if not chunk:
            break
        dst.write(chunk)
        dst.flush()

Использование:

from io import StringIO

src = StringIO("hello\nworld\n")
dst = StringIO()
copy_text(src, dst)
print(dst.getvalue())  # "hello\nworld\n"

Паттерн: duck typing по TextIOBase => тебе не важно, файл это, сетевой поток или буфер в памяти.

Наследование от TextIOBase: текстовый "фильтр" / декоратор

Частый паттерн: берём существующий текстовый поток и добавляем поведение:

  • логирование,
  • префиксы к строкам,
  • фильтрация,
  • ограничение длины,
  • маскирование секретных данных и т.п.
import io

class PrefixWriter(io.TextIOBase):
    def __init__(self, base: io.TextIOBase, prefix: str):
        self._base = base
        self._prefix = prefix

    def writable(self) -> bool:
        return self._base.writable()

    def write(self, s: str) -> int:
        # добавляем префикс в начало каждой строки
        lines = s.splitlines(keepends=True)
        transformed = "".join(self._prefix + line for line in lines)
        return self._base.write(transformed)

    def flush(self):
        self._base.flush()

    def close(self):
        self._base.close()
        super().close()

Использование:

import sys

log = PrefixWriter(sys.stdout, "[INFO] ")
log.write("hello\nworld\n")
## [INFO] hello
## [INFO] world

Текстовый поток поверх бинарного (своего) источника

Классический стек:

  • свой RawIOBase (сокет, шифр, протокол),
  • сверху BufferedReader/Writer,
  • и только затем TextIOWrapper, который уже TextIOBase.
import io
import socket

sock = socket.socket()
sock.connect(("example.com", 80))

raw = sock.makefile("rb", buffering=0)    # или свой RawIOBase
buf = io.BufferedRWPair(raw, raw)
text = io.TextIOWrapper(buf, encoding="utf-8", newline="\n")

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

Паттерн: не трогать байты вручную, а доверить кодировку/декодировку TextIOWrapper.

Собственный текстовый поток поверх BytesIO / байтов

Иногда хочется "отрезать" пользователю доступ к байтам, дать ему только str-интерфейс.

import io

class MemoryText(io.TextIOBase):
    def __init__(self, encoding="utf-8"):
        self._buffer = io.BytesIO()
        self._encoding = encoding

    def writable(self):
        return True

    def readable(self):
        return True

    def write(self, s: str) -> int:
        data = s.encode(self._encoding)
        self._buffer.write(data)
        return len(s)

    def read(self, size: int = -1) -> str:
        data = self._buffer.read(size if size >= 0 else -1)
        return data.decode(self._encoding)

    def seek(self, pos: int, whence: int = 0):
        return self._buffer.seek(pos, whence)

    def tell(self) -> int:
        return self._buffer.tell()

    def getvalue(self) -> str:
        pos = self._buffer.tell()
        self._buffer.seek(0)
        data = self._buffer.read()
        self._buffer.seek(pos)
        return data.decode(self._encoding)

Использование:

mt = MemoryText()
mt.write("привет\n")
mt.write("мир")
mt.seek(0)
print(mt.read())        # "привет\nмир"
print(mt.getvalue())    # "привет\nмир"

Паттерн: TextIOBase как текстовый фасад над байтовым буфером/устройством.

Построчная обработка (readline/iterator) с дополнительной логикой

TextIOBase даёт readline, итерацию по строкам (for line in f:). Можно наследоваться и переопределять readline, оставляя read как есть (или наоборот).

Пример: обёртка, которая отбрасывает слишком длинные строки

class TruncatingTextReader(io.TextIOBase):
    def __init__(self, base: io.TextIOBase, max_len: int):
        self._base = base
        self._max_len = max_len

    def readable(self) -> bool:
        return self._base.readable()

    def readline(self, size: int = -1) -> str:
        line = self._base.readline(size)
        if len(line) > self._max_len:
            return line[:self._max_len] + "...\n"
        return line

    def __iter__(self):
        # чтобы for line in f работал через наше readline
        while True:
            line = self.readline()
            if line == "":
                break
            yield line

Использование:

with open("log.txt", encoding="utf-8") as f:
    trunc = TruncatingTextReader(f, max_len=80)
    for line in trunc:
        print(line, end="")

Логирующий декоратор для текстовых потоков

Классика: оборачиваем TextIOBase и смотрим, что пишется/читается.

class LoggingText(io.TextIOBase):
    def __init__(self, base: io.TextIOBase, name: str = "LOG"):
        self._base = base
        self._name = name

    def writable(self): return self._base.writable()
    def readable(self): return self._base.readable()

    def write(self, s: str) -> int:
        print(f"[{self._name} WRITE]: {s!r}")
        return self._base.write(s)

    def read(self, size: int = -1) -> str:
        s = self._base.read(size)
        print(f"[{self._name} READ]: {s!r}")
        return s

    def readline(self, size: int = -1) -> str:
        s = self._base.readline(size)
        print(f"[{self._name} READLINE]: {s!r}")
        return s

    def flush(self):
        self._base.flush()

    def close(self):
        self._base.close()
        super().close()

Использование:

import sys

log_stdout = LoggingText(sys.stdout, "STDOUT")
log_stdout.write("Hello\n")

Особенности: кодировка, newline, буферизация

Когда ты строишь текс­товый слой над бинарным (обычно через TextIOWrapper), важно:

  • encoding - какая кодировка (UTF-8, cp1251 и т.п.);
  • errors - что делать с "битой" строкой ("strict", "ignore", "replace");
  • newline - как обрабатывать \n, \r\n, \r;
  • line_buffering - сбрасывать ли буфер при \n.

Паттерн для аккуратного текстового файла:

import io

with open("log.txt", "wb") as f_raw:
    buf = io.BufferedWriter(f_raw)
    text = io.TextIOWrapper(
        buf,
        encoding="utf-8",
        newline="\n",
        line_buffering=True,
    )
    text.write("привет\n")
    # flush произойдёт из-за line_buffering

Даже если напрямую TextIOBase ты не наследуешь, понимать этот слой важно, потому что open(..., encoding=...) на самом деле под капотом создаёт TextIOWrapper над Buffered над FileIO.

Когда стоит реально наследоваться от TextIOBase

Имеет смысл, если:

  • уже есть какой-то текстовый источник/приёмник (логгер, GUI widget, сеть), у него есть естественная работа со str, а байты тебе не нужны;
  • нужно сделать текстовый фильтр/декоратор (логирование, ограничение, маскирование), и ты хочешь использовать его везде, где ожидается TextIOBase (например, подменить sys.stdout).

Обычно достаточно обёртки, которая:

  • хранит base: TextIOBase,
  • переопределяет write/read/readline/__iter__/flush/close,
  • делегирует остальное в base.