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 timeimport sysdef 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 subprocessp = 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 subprocessp = 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 subprocessimport timep = 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 timeimport sysimport signalrun = Truedef 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 subprocessp = subprocess.Popen(['python', 'test.py'])wait_count = 1while 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: