io.IOBase - это базовый абстрактный класс для потоков в Python. Напрямую его почти не используют, но он важен как общий протокол для работы с файлами/потоками и как база для своих классов.
Разберем паттерны использования io.IOBase() в Python.
IOBase как общий тип (аннотации и duck-typing)Когда вам нужна функция, работающая с "любыми файловыми объектами" (файлы, BytesIO, StringIO, сокеты, кастомные классы), удобно типизировать параметр как io.IOBase.
from io import IOBase def copy_stream(src: IOBase, dst: IOBase, chunk_size: int = 8192) -> None: while True: chunk = src.read(chunk_size) if not chunk: break dst.write(chunk) dst.flush()
Использование:
from io import BytesIO, StringIO src = StringIO("hello") dst = StringIO() copy_stream(src, dst) print(dst.getvalue()) # "hello"
Вы можете передавать open(...), BytesIO, StringIO, кастомные потоки - всё, что реализует интерфейс IOBase.
IOBase для своих потоковЧаще наследуются от более конкретных классов (RawIOBase, BufferedIOBase, TextIOBase), но иногда достаточно и IOBase, если поток "логический" и сильно кастомный.
Минимальный паттерн - реализовать нужные методы (read, write, seek, tell, close, readable, writable, seekable).
import io class UpperWriter(io.IOBase): def __init__(self, base: io.IOBase): if not base.writable(): raise ValueError("base stream must be writable") self._base = base self._closed = False def writable(self) -> bool: return True def write(self, s): if self._closed: raise ValueError("I/O operation on closed file.") # поддержим и bytes, и str if isinstance(s, str): s = s.upper() elif isinstance(s, (bytes, bytearray)): s = s.upper() else: raise TypeError("str or bytes expected") return self._base.write(s) def close(self): if not self._closed: self._base.close() self._closed = True @property def closed(self): return self._closed
Использование:
import sys up = UpperWriter(sys.stdout) up.write("hello\n") # печатает HELLO
Очень распространённый паттерн: принимаем поток в конструктор и добавляем поведение.
Типичные применения:
class CountingReader(io.IOBase): def __init__(self, base: io.IOBase): if not base.readable(): raise ValueError("base stream must be readable") self._base = base self.bytes_read = 0 def readable(self) -> bool: return True def read(self, size=-1): data = self._base.read(size) if isinstance(data, (bytes, bytearray)): self.bytes_read += len(data) elif isinstance(data, str): # для текстовых потоков self.bytes_read += len(data.encode("utf-8")) return data def close(self): self._base.close()
with) через IOBaseВсе потомки IOBase уже поддерживают протокол контекстного менеджера (__enter__/__exit__), поэтому ваш кастомный класс, если он наследуется от IOBase, автоматически работает с with (при условии, что корректно реализован close()).
class MyStream(io.IOBase): def __init__(self, name): self.name = name self._closed = False def write(self, s): if self._closed: raise ValueError("closed") print(f"[{self.name}] {s}") return len(s) def writable(self): return True def close(self): self._closed = True ## контекстный менеджер: with MyStream("test") as s: s.write("hello") ## при выходе из with будет вызван s.close()
Иногда нужно обернуть "чужой" объект (например, очередь, сокет библиотеки, API HTTP-клиента) в объект, который ведёт себя как файл (read, write, close), чтобы использовать его с чужим кодом.
class GeneratorReader(io.IOBase): def __init__(self, gen): self._gen = gen self._buffer = "" self._closed = False def readable(self): return True def read(self, size=-1): if self._closed: return "" # Простой пример: просто склеиваем всё, что осталось parts = [self._buffer] self._buffer = "" for chunk in self._gen: parts.append(chunk) if size != -1 and sum(len(p) for p in parts) >= size: break data = "".join(parts) if size != -1 and len(data) > size: self._buffer = data[size:] data = data[:size] return data def close(self): self._closed = True
В тестах удобно подсовывать объект, наследованный от IOBase, вместо реального файла или сокета:
read поднимает исключение после N байт)write, потом flush и т.д.)Пример простого "фейкового" текстового файла:
class FakeTextFile(io.IOBase): def __init__(self, initial=""): self._buf = initial self._pos = 0 self._closed = False def readable(self): return True def writable(self): return True def seekable(self): return True def read(self, size=-1): if size == -1: res = self._buf[self._pos:] self._pos = len(self._buf) else: res = self._buf[self._pos:self._pos + size] self._pos += len(res) return res def write(self, s): before = self._buf[:self._pos] after = self._buf[self._pos + len(s):] self._buf = before + s + after self._pos += len(s) return len(s) def seek(self, offset, whence=0): if whence == 0: self._pos = offset elif whence == 1: self._pos += offset elif whence == 2: self._pos = len(self._buf) + offset return self._pos def tell(self): return self._pos
Чаще всего вместо голого IOBase лучше наследоваться от более подходящего базового класса:
io.RawIOBase - низкоуровневые бинарные потоки (readinto, readall, write => bytes)io.BufferedIOBase - буферизованные бинарные (read, read1, readinto, write)io.TextIOBase - текстовые потоки (read, write => str, encoding, newline…)Паттерн такой:
bytes - берите RawIOBase/BufferedIOBaseTextIOBaseIOBasereadable()/writable()/seekable()IOBase даёт стандартный способ узнать, что поток умеет:
def process_stream(f: io.IOBase): if f.readable(): data = f.read() ... if f.writable(): f.write("result\n")
Так можно писать код, который работает с разными типами потоков, не опираясь на конкретные классы.