I created a simple Python function:
import subprocess
from io import TextIOWrapper
def run_shell_command(command: list, debug: bool = False):
'''
Run shell command
:param command: Shell command
:param debug: Debug mode
:return: Result code and message
'''
try:
process = subprocess.run(
command, check=True, text=True, timeout=5,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT
)
if debug:
for line in TextIOWrapper(process.stdout, encoding='utf-8'):
print(line)
message = 'Shell command executed sucessfully'
return ({'code': 200, 'msg': message, 'stdout': process.stdout})
except subprocess.CalledProcessError as e:
return ({'code': 500, 'msg': e.output})
if __name__ == "__main__":
command = run_shell_command(['ls', '-lah'], True)
print(command)
When I run it in debug mode, I get the following error:
Traceback (most recent call last):
File "/tmp/command.py", line 28, in <module>
command = run_shell_command(['ls', '-lah'], True)
File "/tmp/command.py", line 19, in run_shell_command
for line in TextIOWrapper(process.stdout, encoding="utf-8"):
AttributeError: 'str' object has no attribute 'readable'
Running Python 3.9 on a Linux server, I was wondering if you can provide some insight where the issue might be. With debug disabled, I get a proper text output. Thank you for your help.
Edit: Based on the comments below, the fix is quite simple:
if debug:
print(process.stdout.rstrip())
However, IMO the accepted solution is better, compared to original code.
Unfortunately "simple" doesn't cut it with all of the corner cases involved – you'll need to read the subprocess stdout as it's being streamed, print it out and accumulate it in a buffer, and keep track of time so you can timeout correctly. Note that this too has a possible bug (though not very dangerous) if the 4096 byte read happens to end at a multi-line character.
def run_shell_command(command: list, debug: bool = False, timeout: float = 5):
"""
Run shell command
:param command: Shell command
:param debug: Debug mode
:return: Result code and message
"""
with subprocess.Popen(
command,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
) as process:
start_time = time.time()
output = b""
while True:
buf = process.stdout.read(4096)
if debug:
# could fail if `buf` happens to end in a multi-byte character
print(buf.decode("utf-8", "ignore"))
output += buf
if time.time() - start_time > timeout:
process.kill()
message = "Shell command timed out"
return {"code": 500, "msg": message, "stdout": output}
if process.poll() is not None:
break
if process.returncode != 0:
message = "Shell command failed"
return {"code": 500, "msg": message, "stdout": output}
message = "Shell command executed successfully"
return {"code": 200, "msg": message, "stdout": output}
if __name__ == "__main__":
command = run_shell_command(["ls", "-lah"], True)
print(command)