run_command_python.md

import pathlib
ROOT_DIR = pathlib.Path(__file__).parent.parent

process, output = run_process(f"python3 {ROOT_DIR / 'something.py'}", ROOT_DIR)
process = run_background(f"python3 {ROOT_DIR / 'something.py'}", ROOT_DIR)

import os
import sys
import shlex
import subprocess
import time
import signal

def run_process(command_line, directory=None, verbose=False):
    """
    Given a command line as `ls -a` runs this command on the current directory

    https://docs.python.org/3.6/library/subprocess.html#subprocess.Popen
    return (subprocess.POpen object, output_lines as list)
    """
    output_lines = []
    command = shlex.split(command_line)

    if verbose:
        print('run_process command', command, directory, file=sys.stderr)

    with subprocess.Popen(
        command,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        env=os.environ,
        cwd=directory,
    ) as process:
        # https://stackoverflow.com/questions/18421757/live-output-from-subprocess-command
        # https://stackoverflow.com/questions/1606795/catching-stdout-in-realtime-from-subprocess
        while True:
            line = process.stdout.readline()
            if not line:
                break

            line = line.decode("UTF-8", errors='replace')
            line = line.replace('\r\n', '\n').rstrip(' \n\r')
            output_lines.append(line)

            if verbose:
                print(line, file=sys.stderr)
    return process, "\n".join(output_lines)


def run_background(command_line, directory=None, verbose=False):
    command = shlex.split(command_line)
    if verbose:
        print('run_process command', command, directory, file=sys.stderr)

    process = subprocess.Popen(
            command,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            env=os.environ,
            cwd=directory,
        )

    if verbose:
        def back():
            while True:
                line = process.stdout.readline()
                if not line:
                    break
                line = line.decode("UTF-8", errors='replace')
                line = line.replace('\r\n', '\n').rstrip(' \n\r')
                print(line, file=sys.stderr)
        thread = threading.Thread(target=back, daemon=True)
        thread.start()
    return process


def gracifully_kill_process(process, timeout=1, verbose=False):
    process.send_signal(signal.SIGINT)
    time.sleep(timeout)
    if process.poll() is None:
        if verbose:
            print(f"Warning: Could not terminate the process {process.pid} with SIGINT...", file=sys.stderr)
        process.terminate()
        time.sleep(timeout)
        if process.poll() is None:
            if verbose:
                print(f"Warning: Could not terminate the process {process.pid} with SIGTERM...", file=sys.stderr)
            process.kill()
            time.sleep(timeout)
            if process.poll() is None:
                print(
                        f"Warning: Could not terminate the process {process.pid} with SIGKILL, giving up...",
                        file=sys.stderr,
                    )
                return 1
    if verbose:
        print(f"Successfully terminated the process {process.pid}={process.returncode}...", file=sys.stderr)
    return 0