Не закрываем sys.stdout в блоке with: решения на Python

Как избежать закрытия sys.stdout при использовании with в Python: обёртка KeepOpenWrapper, безопасный nullcontext и дублирование дескриптора через os.dup.

Когда вы передаёте sys.stdout (или любой открытый поток) в код, который оборачивает его в блок with, есть риск, что поток закроют у вас из‑под ног. Для временных файлов это нормально, но для потоков, общих для всего процесса, — беда. Задача — перенаправлять байты в файловоподобный объект, не позволяя менеджеру контекста закрыть лежащий в основе дескриптор.

Минимальный пример, воспроизводящий проблему

Предположим, есть универсальная функция, которая потребляет итератор байтовых фрагментов и записывает их в файловоподобный объект. Функция ожидает менеджер контекста и использует with, что и становится источником проблемы, если передать ей sys.stdout.buffer.

import io
from typing import IO

def pump_bytes(target: IO[bytes], source):
    with target as handle:
        for chunk in source:
            handle.write(chunk)

Передача sys.stdout.buffer один раз срабатывает, а при следующем использовании падает, потому что блок with его закрывает:

import sys

buf = sys.stdout.buffer
pump_bytes(buf, io.BytesIO(b"text\n"))  # выводит: text
pump_bytes(buf, io.BytesIO(b"more text\n"))  # ValueError: операция ввода-вывода над закрытым файлом.

Почему так происходит

Блок with запускает протокол управления контекстом: в начале он вызывает __enter__, а в конце — __exit__. Для файловоподобных объектов __exit__ обычно закрывает поток. Поведение по умолчанию — как раз то, чего не хочется для глобальных потоков вроде sys.stdout. Цель — сохранить логику записи, но предотвратить закрытие в штатном (без исключений) сценарии.

Не закрывающая обёртка вокруг менеджеров контекста

Если менять pump_bytes нельзя, проблему решает тонкая обёртка, перехватывающая __exit__. Она проксирует всё к оборачиваемому объекту, кроме выхода без исключений — в этом случае намеренно ничего не делает.

import io
import sys
from typing import IO

class KeepOpenWrapper:
    def __init__(self, inner):
        self._inner = inner

    def __getattr__(self, name):
        return getattr(self._inner, name)

    def __enter__(self):
        return self._inner.__enter__()

    def __exit__(self, exc_type, exc_value, tb):
        if exc_type is not None:
            # В случае исключения передаём очистку оригинальному объекту
            return self._inner.__exit__(exc_type, exc_value, tb)
        # Иначе — не закрываем


def pump_bytes(target: IO[bytes], source):
    with target as handle:
        for chunk in source:
            handle.write(chunk)

nonclosing_stdout = KeepOpenWrapper(sys.stdout.buffer)
pump_bytes(nonclosing_stdout, io.BytesIO(b"text\n"))       # выводит: text
pump_bytes(nonclosing_stdout, io.BytesIO(b"more text\n"))   # выводит: more text

Так sys.stdout.buffer остаётся пригодным для повторных вызовов: закрытие, которое обычно происходит в конце блока with, подавляется.

Важная оговорка: контекстные менеджеры на основе генераторов

Есть случай, когда такая обёртка не поможет. Если объект создаётся через @contextlib.contextmanager, закрытие часто выполняется в блоке finally внутри генератора, и обойти это снаружи нельзя.

import contextlib

class Printer:
    def __init__(self):
        print("Opening Printer")

    def write(self, content):
        print(content)

    def close(self):
        print("Closing Printer")

@contextlib.contextmanager
def managed_printer():
    p = Printer()
    try:
        yield p
    finally:
        # KeepOpenWrapper не сможет помешать выполнению этого кода.
        p.close()

import io

def pump_bytes(target, source):
    with target as handle:
        for chunk in source:
            handle.write(chunk)

pump_bytes(managed_printer(), io.BytesIO(b"some text"))
pump_bytes(KeepOpenWrapper(managed_printer()), io.BytesIO(b"some text"))  # всё равно закрывается

Вывод:

Opening Printer
b'some text'
Closing Printer
Opening Printer
b'some text'
Closing Printer

Менеджер контекста из стандартной библиотеки, который ничего не делает

Когда объект уже является полноценным файловоподобным и его не нужно закрывать, в стандартной библиотеке есть готовое решение: contextlib.nullcontext. Он возвращает объект из __enter__, а его __exit__ ничего не делает — идеальный вариант для обёртки вокруг sys.stdout.buffer.

import contextlib
import io
import sys

cm_stdout = contextlib.nullcontext(sys.stdout.buffer)

def pump_bytes(target, source):
    with target as handle:
        for chunk in source:
            handle.write(chunk)

pump_bytes(cm_stdout, io.BytesIO(b"text\n"))       # выводит: text
pump_bytes(cm_stdout, io.BytesIO(b"more text\n"))   # выводит: more text

Дублирование stdout, чтобы оригинал оставался открытым

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

import io
import os
import sys

fd_clone = os.dup(sys.stdout.fileno())

with os.fdopen(fd_clone, mode="wb") as dup_stream:
    # pump_bytes закроет только дубликат
    pump_bytes(dup_stream, io.BytesIO(b"text\n"))

print("Original stdout still open!")

Почему это важно

Когда вы строите более высокоуровневые I/O-потоки — например, интерфейс наподобие subprocess.run поверх websocket, который гоняет данные к удалённому процессу и обратно, — важно сохранять семантику того, должен ли поток закрываться. Некоторые приёмники, вроде open("file"), нужно закрывать. Другие, такие как sys.stdout или дескриптор 1, должны оставаться открытыми. Возможность передавать это намерение через слои без дополнительных флагов делает дизайн чище и предсказуемее.

Практические выводы

Если вызываемая функция использует with для переданного файловоподобного объекта, а вам нужно предотвратить закрытие, оберните объект менеджером контекста, который ничего не делает при штатном выходе. contextlib.nullcontext — самый простой вариант, когда вы передаёте существующий поток вроде sys.stdout.buffer. Если требуется отдельная закрываемая ручка, не влияющая на исходный поток, продублируйте дескриптор через os.dup и работайте как обычно. Помните: контекстные менеджеры на основе генераторов, которые делают очистку в finally, подавить снаружи нельзя — в таких случаях поведение с закрытием является частью контракта менеджера.