Add system_raise_on_error config option for ! shell operator (#15073)

Carreau · web-flow · commit cf9f2b27790c · 2026-02-19T10:21:08.000+01:00
Fixes #12264 Add system_raise_on_error configuration option that raises CalledProcessError when shell commands executed via the ! operator return non-zero exit status. This brings similar error-handling behavior to the ! operator as was added to %%bash magic in PR #11287. Changes: - Add system_raise_on_error Bool traitlet config (default: False) - Import CalledProcessError from subprocess - Modify system_piped() to raise on non-zero exit when enabled - Modify system_raw() to raise on non-zero exit when enabled - Modify getoutput() to raise on non-zero exit when enabled (uses get_output_error_code to capture exit status) When enabled, users can halt notebook execution on command failures: get_ipython().system_raise_on_error = True !false # Now raises CalledProcessError Note: A follow-up PR to ipykernel will be needed to support this in Jupyter notebooks, as ZMQInteractiveShell overrides system_piped().
diff --git a/IPython/core/interactiveshell.py b/IPython/core/interactiveshell.py
@@ -22,6 +22,7 @@
 import runpy
 import shutil
 import subprocess
+from subprocess import CalledProcessError
 import sys
 import tempfile
 import traceback
@@ -90,7 +91,7 @@
 from IPython.utils.io import ask_yes_no
 from IPython.utils.ipstruct import Struct
 from IPython.utils.path import ensure_dir_exists, get_home_dir, get_py_filename
-from IPython.utils.process import getoutput, system
+from IPython.utils.process import get_output_error_code, getoutput, system
 from IPython.utils.strdispatch import StrDispatch
 from IPython.utils.syspathcontext import prepended_to_syspath
 from IPython.utils.text import DollarFormatter, LSString, SList, format_screen
@@ -538,6 +539,14 @@ def input_transformers_cleanup(self):
 
     quiet = Bool(False).tag(config=True)
 
+    system_raise_on_error = Bool(False, help=
+        """
+        Raise an exception on non-zero exit status from shell commands executed
+        via the `!` operator. When set to True, shell commands that fail will raise
+        CalledProcessError, similar to the behavior of %%script magics.
+        """
+    ).tag(config=True)
+
     history_length = Integer(10000,
         help='Total length of command history'
     ).tag(config=True)
@@ -2646,7 +2655,12 @@ def system_piped(self, cmd):
         # we explicitly do NOT return the subprocess status code, because
         # a non-None value would trigger :func:`sys.displayhook` calls.
         # Instead, we store the exit_code in user_ns.
-        self.user_ns['_exit_code'] = system(self.var_expand(cmd, depth=1))
+        exit_code = system(self.var_expand(cmd, depth=1))
+        self.user_ns['_exit_code'] = exit_code
+
+        # Raise an exception if the command failed and system_raise_on_error is True
+        if self.system_raise_on_error and exit_code != 0:
+            raise CalledProcessError(exit_code, cmd)
 
     def system_raw(self, cmd):
         """Call the given cmd in a subprocess using os.system on Windows or
@@ -2712,6 +2726,10 @@ def system_raw(self, cmd):
         # but raising SystemExit(_exit_code) will give status 254!
         self.user_ns['_exit_code'] = ec
 
+        # Raise an exception if the command failed and system_raise_on_error is True
+        if self.system_raise_on_error and ec != 0:
+            raise CalledProcessError(ec, cmd)
+
     # use piped system by default, because it is better behaved
     system = system_piped
 
@@ -2737,11 +2755,27 @@ def getoutput(self, cmd, split=True, depth=0):
         if cmd.rstrip().endswith('&'):
             # this is *far* from a rigorous test
             raise OSError("Background processes not supported.")
-        out = getoutput(self.var_expand(cmd, depth=depth+1))
+
+        # Get output and exit code
+        expanded_cmd = self.var_expand(cmd, depth=depth+1)
+        if self.system_raise_on_error:
+            # Use get_output_error_code to get the exit code
+            out_str, err_str, exit_code = get_output_error_code(expanded_cmd)
+            # Combine stdout and stderr as getoutput does
+            out_combined = out_str if not err_str else out_str + err_str
+            self.user_ns['_exit_code'] = exit_code
+
+            # Raise an exception if the command failed
+            if exit_code != 0:
+                raise CalledProcessError(exit_code, cmd)
+        else:
+            # Use the original getoutput for backward compatibility
+            out_combined = getoutput(expanded_cmd)
+
         if split:
-            out = SList(out.splitlines())
+            out = SList(out_combined.splitlines())
         else:
-            out = LSString(out)
+            out = LSString(out_combined)
         return out
 
     #-------------------------------------------------------------------------
