Библиотека io в Python для ввода и вывода

Расширенные примеры использования библиотеки io

Пример 1: Создание собственного потока на основе RawIOBase

Наследуясь от io.RawIOBase, можно реализовать поток, который читает данные из нестандартного источника (например, базы данных или генератора).

import io

class SimpleStream(io.RawIOBase):
    def __init__(self, data):
        self.data = data
        self.pos = 0

    def readable(self):
        return True

    def read(self, size=-1):
        if size < 0:
            rest = self.data[self.pos:]
            self.pos = len(self.data)
            return rest
        else:
            chunk = self.data[self.pos:self.pos+size]
            self.pos += len(chunk)
            return chunk

    def readinto(self, b):
        chunk = self.read(len(b))
        b[:len(chunk)] = chunk
        return len(chunk)

raw = SimpleStream(b'Testing raw stream')
buffered = io.BufferedReader(raw)
print(buffered.read(7))
print(buffered.read())

b'Testing'
b' raw stream'

Пример 2: Работа с архивами в памяти (gzip + BytesIO)

Сжатие и распаковка данных напрямую в байтовом буфере без создания файлов.

import io
import gzip

# Запись сжатых данных
buf = io.BytesIO()
with gzip.GzipFile(fileobj=buf, mode='wb') as gz:
    gz.write(b'Compressed content')

# Чтение сжатых данных
buf.seek(0)
with gzip.GzipFile(fileobj=buf, mode='rb') as gz:
    decompressed = gz.read()
print(decompressed)

b'Compressed content'

Пример 3: Использование TextIOWrapper для обработки разных кодировок

Чтение данных, закодированных в CP1251, с заменой неподдерживаемых символов.

import io

raw = io.BytesIO(b'\xcf\xf0\xee\xe2\xe5\xf0\xea\xe0')
wrapper = io.TextIOWrapper(raw, encoding='cp1251', errors='replace')
print(wrapper.readline())

Проверка

Пример 4: Комбинация BufferedRandom для произвольного доступа

io.BufferedRandom позволяет одновременно читать и писать в бинарный поток.

import io

buf = io.BytesIO(b'Hello World')
buffered = io.BufferedRandom(buf)
buffered.seek(6)
buffered.write(b'Python')
buffered.seek(0)
print(buffered.read())

b'Hello Python'

Пример 5: Использование io.open для явного управления буферизацией

Функция io.open() позволяет задать размер буфера и политику кэширования.

import io

with io.open('test.txt', 'w', buffering=1024*1024) as f:
    f.write('Large buffer test')

with io.open('test.txt', 'r', buffering=1) as f:
    print(f.read())

Large buffer test

Пример 6: Поток с возможностью повторного чтения (io.BufferedReader с позицией)

Сброс позиции после чтения – использование seek(0) для повторного чтения.

import io

data = b'Repeatable'
buf = io.BytesIO(data)
buff = io.BufferedReader(buf)
print(buff.read())
buff.seek(0)
print(buff.read())

b'Repeatable'
b'Repeatable'

Пример 7: Обработка ошибок кодировки при записи через StringIO

При записи в текстовый поток можно указать стратегию обработки ошибок (строгий режим, замена, игнорирование).

import io

# Пример с ошибкой – символ вне Latin-1
buf = io.StringIO()
try:
    buf.write('\u2603')  # снеговик вне Latin-1
    # Если поток имеет кодировку ascii, произойдёт ошибка
    # Но StringIO не привязан к кодировке, поэтому ошибки нет
except Exception as e:
    print(f'Ошибка: {e}')

# Для BytesIO с TextIOWrapper можно задать errors
raw = io.BytesIO()
wrapper = io.TextIOWrapper(raw, encoding='ascii', errors='replace')
wrapper.write('\u2603')
wrapper.flush()
print(raw.getvalue())

b'?'

Пример 8: Поток с поддержкой seek на основе RandomAccessIO

Использование io.BytesIO для быстрого перемещения по большому объёму данных, например, для поиска подстроки.

import io

big_data = b'x' * 10000 + b'needle' + b'y' * 10000
buf = io.BytesIO(big_data)
# Поиск через чтение порциями
while True:
    chunk = buf.read(1024)
    if not chunk:
        break
    pos = chunk.find(b'needle')
    if pos != -1:
        print(f'Found at position {buf.tell() - len(chunk) + pos}')
        break

Found at position 10000