Function _system_body

IPython/utils/_process_win32.py:68–113

Callback for _system.

(p: subprocess.Popen[bytes])

def _system_body(p: subprocess.Popen[bytes]) -> int:
    """Callback for _system."""
    enc = DEFAULT_ENCODING

    # Dec 2024: in both of these functions, I'm not sure why we .splitlines()
    # the bytes and then decode each line individually instead of just decoding
    # the whole thing at once.
    def stdout_read() -> None:
        try:
            assert p.stdout is not None
            for byte_line in (read_no_interrupt(p.stdout) or b"").splitlines():
                line = byte_line.decode(enc, "replace")
                print(line, file=sys.stdout)
        except Exception as e:
            print(f"Error reading stdout: {e}", file=sys.stderr)

    def stderr_read() -> None:
        try:
            assert p.stderr is not None
            for byte_line in (read_no_interrupt(p.stderr) or b"").splitlines():
                line = byte_line.decode(enc, "replace")
                print(line, file=sys.stderr)
        except Exception as e:
            print(f"Error reading stderr: {e}", file=sys.stderr)

    stdout_thread = Thread(target=stdout_read)
    stderr_thread = Thread(target=stderr_read)

    stdout_thread.start()
    stderr_thread.start()

    # Wait to finish for returncode. Unfortunately, Python has a bug where
    # wait() isn't interruptible (https://bugs.python.org/issue28168) so poll in
    # a loop instead of just doing `return p.wait()`
    while True:
        result = p.poll()
        if result is None:
            time.sleep(0.01)
        else:
            break

    # Join the threads to ensure they complete before returning
    stdout_thread.join()
    stderr_thread.join()

    return result


def system(cmd: str) -> Optional[int]: