Simple Guide to Subprocess (launch another python script)

December 9, 2020

Note

If you need your python code to run in multiple process/CPU, you should use multiprocessing.

If somehow you code doesn’t work well using multiprocessing or it is too much effort, you can launch your code as python script.

Create a python script

Create test.py

import time
import sys

def main():
    loop_count = 0
    while True:
        print('.', end='', flush=True)
        loop_count += 1
        if loop_count > 10:
            break
        #if loop_count > 5:
        #    raise Exception('TEST')
        time.sleep(1)

    print('END')
    sys.exit(0)


if __name__ == '__main__':
    main()

subprocess.run

This is simpler, but provide less control.

Launch a process to execute the python script, and wait until the process to complete.

NOTE: You will notice the output of script is printed on your console.

import subprocess
p = subprocess.run(['python', 'test.py'])
# or
# p = subprocess.run('python test.py', shell=True)

print('returncode', p.returncode)
print('EXIT')

To capture the output.

NOTE: Will raise subprocess.CalledProcessError: Command 'xxx' returned non-zero exit status 2. if the script has an exception.

output = subprocess.check_output(['python', 'test.py'])
print(output)

subprocess.Popen

Launch a process to execute the python script. This will not wait for the script to complete.

NOTE: You will notice the output of script is printed on your console.

import subprocess

p = subprocess.Popen(['python', 'test.py'])
print('pid', p.pid)
print('EXIT')

Capture output and error, and will wait for process to end.

p = subprocess.Popen(['python', 'test.py'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
out, err = p.communicate()
print('out', out)
print('err', err)
print('returncode', p.returncode)
print('EXIT')

Write output and error to file, but don’t want for process to end.

with open("test.log","wb") as out, open("test-error.log","wb") as err:
    p = subprocess.Popen(['python', 'test.py'], stdout=out, stderr=err)

Poll until process end.

import subprocess
import time

p = subprocess.Popen(['python', 'test.py'])

while p.poll() is None:
    print('.', end='', flush=True)
    time.sleep(1)

print('returncode', p.returncode)

Wait until process end.

p = subprocess.Popen(['python', 'test.py'])
p.wait()
print('returncode', p.returncode)

Send Signal for graceful termination

test.py

import time
import sys
import signal

run = True

def stop(signum, frame):
    global run
    print('STOP')
    run = False
    print('STOP2')

signal.signal(signal.SIGTERM, stop)

def main():
    loop_count = 0
    while run:
        print('.', end='', flush=True)
        loop_count += 1
        if loop_count > 10:
            break
        #if loop_count > 5:
        #    raise Exception('TEST')
        time.sleep(1)

    print('END')
    sys.exit(0)


if __name__ == '__main__':
    main()

main.py

import subprocess

p = subprocess.Popen(['python', 'test.py'])
wait_count = 1
while p.poll() is None:
    # print('.', end='', flush=True)
    wait_count += 1
    if wait_count > 3:
        print('Send TERMINATE')
        p.terminate()
    time.sleep(1)

print('returncode', p.returncode)

References:

This work is licensed under a
Creative Commons Attribution-NonCommercial 4.0 International License.