Function process_handler

IPython/utils/_process_common.py:48–121

Open a command in a shell subprocess and execute a callback. This function provides common scaffolding for creating subprocess.Popen() calls. It creates a Popen object and then calls the callback with it. Parameters ---------- cmd : str or list A command to be executed

(
    cmd: Union[str, List[str]],
    callback: Callable[[subprocess.Popen[bytes]], _T],
    stderr: int = subprocess.PIPE,
)

def process_handler(
    cmd: Union[str, List[str]],
    callback: Callable[[subprocess.Popen[bytes]], _T],
    stderr: int = subprocess.PIPE,
) -> _T | None:
    """Open a command in a shell subprocess and execute a callback.

    This function provides common scaffolding for creating subprocess.Popen()
    calls.  It creates a Popen object and then calls the callback with it.

    Parameters
    ----------
    cmd : str or list
        A command to be executed by the system, using :class:`subprocess.Popen`.
        If a string is passed, it will be run in the system shell. If a list is
        passed, it will be used directly as arguments.
    callback : callable
        A one-argument function that will be called with the Popen object.
    stderr : file descriptor number, optional
        By default this is set to ``subprocess.PIPE``, but you can also pass the
        value ``subprocess.STDOUT`` to force the subprocess' stderr to go into
        the same file descriptor as its stdout.  This is useful to read stdout
        and stderr combined in the order they are generated.

    Returns
    -------
    The return value of the provided callback is returned.
    """
    sys.stdout.flush()
    sys.stderr.flush()
    # On win32, close_fds can't be true when using pipes for stdin/out/err
    if sys.platform == "win32" and stderr != subprocess.PIPE:
        close_fds = False
    else:
        close_fds = True
    # Determine if cmd should be run with system shell.
    shell = isinstance(cmd, str)
    # On POSIX systems run shell commands with user-preferred shell.
    executable = None
    if shell and os.name == 'posix' and 'SHELL' in os.environ:
        executable = os.environ['SHELL']
    p = subprocess.Popen(cmd, shell=shell,
                         executable=executable,
                         stdin=subprocess.PIPE,
                         stdout=subprocess.PIPE,
                         stderr=stderr,
                         close_fds=close_fds)

    try:
        out = callback(p)
    except KeyboardInterrupt:
        print('^C')
        sys.stdout.flush()
        sys.stderr.flush()
        out = None
    finally:
        # Make really sure that we don't leave processes behind, in case the
        # call above raises an exception