python_fix_brokenpipeerror.py

#!/usr/bin/env python

# https://adamj.eu/tech/2025/07/20/python-fix-brokenpipeerror/

import os
import sys
from contextlib import contextmanager
from typing import Generator


@contextmanager
def handle_broken_pipe() -> Generator[None]:
    """
    Prevent BrokenPipeError when the output stream is closed early, such as
    when piping to head.
    """
    try:
        yield
        sys.stdout.flush()
    except BrokenPipeError:
        # Python flushes standard streams on exit; redirect remaining output
        # to devnull to avoid another BrokenPipeError at shutdown
        devnull = os.open(os.devnull, os.O_WRONLY)
        os.dup2(devnull, sys.stdout.fileno())


with handle_broken_pipe():
    for i in range(1_000):
        print(f"This is line {i}")