handle broken pipe

CHANGELOG.md

@@ -18,6 +18,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Fixed cached hash preservation upon clearing meta and links https://github.com/Textualize/rich/issues/2942
 - Fixed overriding the `background_color` of `Syntax` not including padding https://github.com/Textualize/rich/issues/3295
 - Fixed selective enabling of highlighting when disabled in the `Console` https://github.com/Textualize/rich/issues/3419
+- Fixed BrokenPipeError writing an error message https://github.com/Textualize/rich/pull/3468
 
 ### Changed
 
@@ -31,6 +32,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Adds a `case_sensitive` parameter to `prompt.Prompt`. This determines if the
   response is treated as case-sensitive. Defaults to `True`.
+- Added `Console.on_broken_pipe` https://github.com/Textualize/rich/pull/3468
 
 ## [13.7.1] - 2024-02-28
 

rich/console.py

@@ -1385,9 +1385,14 @@ def render_lines(
                 extra_lines = render_options.height - len(lines)
                 if extra_lines > 0:
                     pad_line = [
-                        [Segment(" " * render_options.max_width, style), Segment("\n")]
-                        if new_lines
-                        else [Segment(" " * render_options.max_width, style)]
+                        (
+                            [
+                                Segment(" " * render_options.max_width, style),
+                                Segment("\n"),
+                            ]
+                            if new_lines
+                            else [Segment(" " * render_options.max_width, style)]
+                        )
                     ]
                     lines.extend(pad_line * extra_lines)
 
@@ -1436,9 +1441,11 @@ def render_str(
             rich_text.overflow = overflow
         else:
             rich_text = Text(
-                _emoji_replace(text, default_variant=self._emoji_variant)
-                if emoji_enabled
-                else text,
+                (
+                    _emoji_replace(text, default_variant=self._emoji_variant)
+                    if emoji_enabled
+                    else text
+                ),
                 justify=justify,
                 overflow=overflow,
                 style=style,
@@ -1989,6 +1996,20 @@ def log(
             ):
                 buffer_extend(line)
 
+    def on_broken_pipe(self) -> None:
+        """This function is called when a `BrokenPipeError` is raised.
+
+        This can occur when piping Textual output in Linux and macOS.
+        The default implementation is to exit the app, but you could implement
+        this method in a subclass to change the behavior.
+
+        See https://docs.python.org/3/library/signal.html#note-on-sigpipe for details.
+        """
+        self.quiet = True
+        devnull = os.open(os.devnull, os.O_WRONLY)
+        os.dup2(devnull, sys.stdout.fileno())
+        raise SystemExit(1)
+
     def _check_buffer(self) -> None:
         """Check if the buffer may be rendered. Render it if it can (e.g. Console.quiet is False)
         Rendering is supported on Windows, Unix and Jupyter environments. For
@@ -1998,6 +2019,15 @@ def _check_buffer(self) -> None:
         if self.quiet:
             del self._buffer[:]
             return
+
+        try:
+            self._write_buffer()
+        except BrokenPipeError:
+            self.on_broken_pipe()
+
+    def _write_buffer(self) -> None:
+        """Write the buffer to the output file."""
+
         with self._lock:
             if self.record:
                 with self._record_buffer_lock:

tests/test_console.py

@@ -1,6 +1,7 @@
 import datetime
 import io
 import os
+import subprocess
 import sys
 import tempfile
 from typing import Optional, Tuple, Type, Union
@@ -1017,3 +1018,23 @@ def test_reenable_highlighting() -> None:
         lines[1]
         == "\x1b[1m[\x1b[0m\x1b[1;36m1\x1b[0m, \x1b[1;36m2\x1b[0m, \x1b[1;36m3\x1b[0m\x1b[1m]\x1b[0m"
     )
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="does not run on windows")
+def test_brokenpipeerror() -> None:
+    """Test BrokenPipe works as expected."""
+    which_py, which_head = (["which", cmd] for cmd in ("python", "head"))
+    rich_cmd = "python -m rich".split()
+    for cmd in [which_py, which_head, rich_cmd]:
+        check = subprocess.run(cmd).returncode
+        if check != 0:
+            return  # Only test on suitable Unix platforms
+    head_cmd = "head -1".split()
+    proc1 = subprocess.Popen(rich_cmd, stdout=subprocess.PIPE)
+    proc2 = subprocess.Popen(head_cmd, stdin=proc1.stdout, stdout=subprocess.PIPE)
+    proc1.stdout.close()
+    output, _ = proc2.communicate()
+    proc1.wait()
+    proc2.wait()
+    assert proc1.returncode == 1
+    assert proc2.returncode == 0