How do I call an external command within Python as if I had typed it in a shell or command prompt?
66 Answers
Here there are a lot of answers, but none fulfilled all my needs.
- I need to run the command and capture the output and exit code.
- I need to timeout the executed program and force it to exit if timeout is reached, and kill all its child processes.
- and I need that it works in Windows XP and later, Cygwin and Linux. In Python 2 and 3.
So I created this:
def _run(command, timeout_s=False, shell=False):
### run a process, capture the output and wait for it to finish. if timeout is specified then Kill the subprocess and its children when the timeout is reached (if parent did not detach)
## usage: _run(arg1, arg2, arg3)
# arg1: command + arguments. Always pass a string; the function will split it when needed
# arg2: (optional) timeout in seconds before force killing
# arg3: (optional) shell usage. default shell=False
## return: a list containing: exit code, output, and if timeout was reached or not
# - Tested on Python 2 and 3 on Windows XP, Windows 7, Cygwin and Linux.
# - preexec_fn=os.setsid (py2) is equivalent to start_new_session (py3) (works on Linux only), in Windows and Cygwin we use TASKKILL
# - we use stderr=subprocess.STDOUT to merge standard error and standard output
import sys, subprocess, os, signal, shlex, time
def _runPY3(command, timeout_s=None, shell=False):
# py3.3+ because: timeout was added to communicate() in py3.3.
new_session=False
if sys.platform.startswith('linux'): new_session=True
p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, start_new_session=new_session, shell=shell)
try:
out = p.communicate(timeout=timeout_s)[0].decode('utf-8')
is_timeout_reached = False
except subprocess.TimeoutExpired:
print('Timeout reached: Killing the whole process group...')
killAll(p.pid)
out = p.communicate()[0].decode('utf-8')
is_timeout_reached = True
return p.returncode, out, is_timeout_reached
def _runPY2(command, timeout_s=0, shell=False):
preexec=None
if sys.platform.startswith('linux'): preexec=os.setsid
p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, preexec_fn=preexec, shell=shell)
start_time = time.time()
is_timeout_reached = False
while timeout_s and p.poll() == None:
if time.time()-start_time >= timeout_s:
print('Timeout reached: Killing the whole process group...')
killAll(p.pid)
is_timeout_reached = True
break
time.sleep(1)
out = p.communicate()[0].decode('utf-8')
return p.returncode, out, is_timeout_reached
def killAll(ParentPid):
if sys.platform.startswith('linux'):
os.killpg(os.getpgid(ParentPid), signal.SIGTERM)
elif sys.platform.startswith('cygwin'):
# subprocess.Popen(shlex.split('bash -c "TASKKILL /F /PID $(</proc/{pid}/winpid) /T"'.format(pid=ParentPid)))
winpid=int(open("/proc/{pid}/winpid".format(pid=ParentPid)).read())
subprocess.Popen(['TASKKILL', '/F', '/PID', str(winpid), '/T'])
elif sys.platform.startswith('win32'):
subprocess.Popen(['TASKKILL', '/F', '/PID', str(ParentPid), '/T'])
# - In Windows, we never need to split the command, but in Cygwin and Linux we need to split if shell=False (default), shlex will split the command for us
if shell==False and (sys.platform.startswith('cygwin') or sys.platform.startswith('linux')):
command=shlex.split(command)
if sys.version_info >= (3, 3): # py3.3+
if timeout_s==False:
returnCode, output, is_timeout_reached = _runPY3(command, timeout_s=None, shell=shell)
else:
returnCode, output, is_timeout_reached = _runPY3(command, timeout_s=timeout_s, shell=shell)
else: # Python 2 and up to 3.2
if timeout_s==False:
returnCode, output, is_timeout_reached = _runPY2(command, timeout_s=0, shell=shell)
else:
returnCode, output, is_timeout_reached = _runPY2(command, timeout_s=timeout_s, shell=shell)
return returnCode, output, is_timeout_reached
Then use it like this:
Always pass the command as one string (it is easier). You do not need to split it; the function will split it when needed.
If your command works in your shell, it will work with this function, so test your command in your shell first cmd/Bash.
So we can use it like this with a timeout:
a=_run('cmd /c echo 11111 & echo 22222 & calc',3)
for i in a[1].splitlines(): print(i)
Or without a timeout:
b=_run('cmd /c echo 11111 & echo 22222 & calc')
More examples:
b=_run('''wmic nic where 'NetConnectionID="Local Area Connection"' get NetConnectionStatus /value''')
print(b)
c=_run('cmd /C netsh interface ip show address "Local Area Connection"')
print(c)
d=_run('printf "<%s>\n" "{foo}"')
print(d)
You can also specify shell=True, but it is useless in most cases with this function. I prefer to choose myself the shell I want, but here it is if you need it too:
# windows
e=_run('echo 11111 & echo 22222 & calc',3, shell=True)
print(e)
# Cygwin/Linux:
f=_run('printf "<%s>\n" "{foo}"', shell=True)
print(f)
Why did I not use the simpler new method subprocess.run()
?
- because it is supported in Python 3.7+, but the last supported Python version in Windows XP is 3.4.
- and because the timeout argument of this function is useless in Windows, it does not kill the child processes of the executed command.
- if you use the
capture_output
+timeout
argument, it will hang if there is a child process still running. And it is still broken in Windows, for which the issue 31447 is still open.
-
1I removed the "2022 answer" because it’s misleading. In 2022 you don’t support winxp nor Python 2 and you use
subprocess.run()
. The question is not "how do I run system commands on old platforms" but "how do I run system commands", period. Commented May 24, 2022 at 10:08
I have written a wrapper to handle errors and redirecting output and other stuff.
import shlex
import psutil
import subprocess
def call_cmd(cmd, stdout=sys.stdout, quiet=False, shell=False, raise_exceptions=True, use_shlex=True, timeout=None):
"""Exec command by command line like 'ln -ls "/var/log"'
"""
if not quiet:
print("Run %s", str(cmd))
if use_shlex and isinstance(cmd, (str, unicode)):
cmd = shlex.split(cmd)
if timeout is None:
process = subprocess.Popen(cmd, stdout=stdout, stderr=sys.stderr, shell=shell)
retcode = process.wait()
else:
process = subprocess.Popen(cmd, stdout=stdout, stderr=sys.stderr, shell=shell)
p = psutil.Process(process.pid)
finish, alive = psutil.wait_procs([p], timeout)
if len(alive) > 0:
ps = p.children()
ps.insert(0, p)
print('waiting for timeout again due to child process check')
finish, alive = psutil.wait_procs(ps, 0)
if len(alive) > 0:
print('process {} will be killed'.format([p.pid for p in alive]))
for p in alive:
p.kill()
if raise_exceptions:
print('External program timeout at {} {}'.format(timeout, cmd))
raise CalledProcessTimeout(1, cmd)
retcode = process.wait()
if retcode and raise_exceptions:
print("External program failed %s", str(cmd))
raise subprocess.CalledProcessError(retcode, cmd)
You can call it like this:
cmd = 'ln -ls "/var/log"'
stdout = 'out.txt'
call_cmd(cmd, stdout)
Sultan is a recent-ish package meant for this purpose. It provides some niceties around managing user privileges and adding helpful error messages.
from sultan.api import Sultan
with Sultan.load(sudo=True, hostname="myserver.com") as sultan:
sultan.yum("install -y tree").run()
Using the Python subprocess module to execute shell commands and write the output to a file.
The below script will run the ps -ef command, filter lines containing python3, and write them to a file called python_processes.txt. Note that the code does not handle any exceptions that might occur during execution.
import subprocess
# Command to execute
cmd = ["ps", "-ef"]
# Execute the command
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
output, error = process.communicate()
# Check if the command was executed without errors
if error is None:
# Filter lines with 'python3'
python_processes = [line for line in output.decode('utf-8').split('\n') if 'python3' in line]
# Write the output to a file
with open('python_processes.txt', 'w') as f:
for process in python_processes:
f.write(process + '\n')
else:
print(f"Error occurred while executing command: {error}")
Here is a Python script that will run the command on Ubuntu, while also showing the logs in real-time:
command = 'your command here'
process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
while True:
output = process.stdout.readline().decode()
if output == '' and process.poll() is not None:
break
if output:
print(output.strip())
rc = process.poll()
if rc == 0:
print("Command succeeded.")
else:
print("Command failed.")
I use this for Python 3.6+:
import subprocess
def execute(cmd):
"""
Purpose : To execute a command and return exit status
Argument : cmd - command to execute
Return : result, exit_code
"""
process = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(result, error) = process.communicate()
rc = process.wait()
if rc != 0:
print ("Error: failed to execute command: ", cmd)
print (error.rstrip().decode("utf-8"))
return result.rstrip().decode("utf-8"), serror.rstrip().decode("utf-8")
# def
-
1Don't use set
shell=True
to run commands, it opens the program to command injection vulnerabilities. You're supposed to pass the command as a list with argumentscmd=["/bin/echo", "hello word"]
. docs.python.org/3/library/… Commented Apr 19, 2020 at 16:52