I'm creating a Python terminal on a Windows computer from scratch and I'm trying to sync the error commands and pipe them from either command prompt or Powershell which ever one the code is in. However, when I use subprocess.run, certain longer commands like netstat don't work. However, when I use subprocess.Popen, the error messages don't route from command prompt correctly.
This is the current code:
def run_os_command():
while True:
cwd = os.getcwd()
current_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
command = input(f"{cwd} {current_time} >>:>> ")
if command.lower() == "exit":
break
if command.lower().startswith("cd"):
try:
path = command.split(' ', 1)[1] # get the path from the command
os.chdir(path)
except (IndexError, FileNotFoundError, NotADirectoryError):
print("Error: Invalid directory")
else:
command = 'powershell ' + command
command = command.split()
try:
process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
while True:
output = process.stdout.readline()
if output == '' and process.poll() is not None:
break
if output:
print(output.strip())
process.poll()
except FileNotFoundError:
print("Error: Command does not exist")
Tried with subprocess.Popen
:>> l
Expected:
"L" is not recognized as an external or batch command....
Obtained: the next prompt
Tried with subprocess.Run
:>> netstat
Expected:
Active Connections
Proto Local Address Foreign Address State
Obtained: nothing, the command lagged indefinitely
subprocess.run()
is a convenience function which first calls subprocess.Popen()
to create a subprocess, then Popen.communicate()
to wait for it to finish and return the return code, stdout
data and stderr
data.
Whereas with calling subprocess.Popen()
, you are only launching the subprocess and the rest is up to you. Program execution continues as soon as the process is created. You can then call Poepen.wait()
to wait for completion, call Popen.communicate()
to optionally send input to the subprocess and wait for completion, or write asynchronous code (by reading from stdout
/stderr
buffers as data is available, using poll()
to determine if subprocess has completed, implementing multithreading, etc). Refer to subprocess library documentation.
The choice comes down to your needs, both methods will work. Now onto your code...
The reason you aren't getting a FileNotFound
exception for invalid commands is when you create the subprocess, you are passing it args=['powershell', 'asdflkj']
--and 'powershell'
IS a valid command. subprocess
parses args
where the first parameter is the command/application, and the rest of the list are arguments for that specific application/command.
This means you need to wait for the subprocess to complete and check the return code of the CompletedSubprocess
. Any code other than 0 indicates failure. Error text will be in process.stderr
But furthermore, you don't even need to launch powershell
; commands can be called directly. I did some quick benchmarks and it is significantly faster on my machine to call commands directly, both when valid and invalid.
One caveat: if you want to be able to run commands which are built in to the shell such as dir
, it requires launching a shell. For this you could use shell=True
argument for Popen()
/run()
rather than calling powershell
yourself. (Online you will find people recommending avoiding shell=True
for security reasons, which I'm assuming you don't need to worry about).
Here is an implementation using run()
:
from datetime import datetime
import os
import subprocess
while True:
cwd = os.getcwd()
command = input(f"{cwd} {datetime.now(): %Y-%m-%d %H:%M:%S} >>:>> ")
if command.strip() == '':
continue
if command.lower() == 'exit':
break
cmd, *args = command.split()
if cmd.lower() == 'cd':
if not args:
print(cwd) # mirrors functionality of actual cd
continue
try:
path = command.split(maxsplit=1)[1] # get the path from the command
os.chdir(path)
except (FileNotFoundError, NotADirectoryError):
print('Error: Invalid directory')
continue
process = subprocess.run([cmd, *args],
stderr=subprocess.PIPE,
text=True,
shell=True)
if process.returncode != 0:
print('Error: Command does not exist')
This approach does not pipe stdout
, meaning output is printed in real-time but you can still display your own error message for invalid commands!
A few fixes/improvements:
cd
without arguments echoes current directory instead of raising uncaught exceptionIf you do want to process/filter the output line-by-line in real time, you can use Popen()
with piped stdout
as follows:
process = subprocess.Popen([cmd, *args],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
shell=True)
for line in process.stdout:
# do something with output line
print(line.strip())
try:
process.wait(timeout=10)
if process.returncode != 0:
print('Error: Command does not exist')
except subprocess.TimeoutExpired:
print('Error: command timed out')
Other notes...
CTRL+C
, you can wrap the loop in a try
block and catch KeyboardInterrupt
subprocess
/Popen
, make sure you're looking at recent info--there is a lot of 10-year-old advice from when the subprocess
library had bugs and lacked current functionality