Python subprocess: Difference between revisions
(→Basics) |
|||
Line 44: | Line 44: | ||
</syntaxhighlight> | </syntaxhighlight> | ||
</blockquote><!-- Windows --> | </blockquote><!-- Windows --> | ||
== wrap stdin/stdout/stderr in subprocess == | |||
<blockquote> | |||
<source lang="python"> | |||
#!/usr/bin/env python | |||
import enum | |||
import os | |||
import re | |||
import select | |||
import subprocess | |||
import sys | |||
class MiddlewareBase: | |||
""" Base Class for middleware | |||
""" | |||
def apply(self, text): | |||
raise NotImplementedError() | |||
class WrapSubprocess: | |||
def wrap(self, pipe, stdout_middleware=None, stderr_middleware=None): | |||
""" Wraps a subprocess.Popen(), allowing you to modify it's stdin/stdout/stderr with lists of 'middleware' functions. | |||
""" | |||
while True: | |||
# exit when subprocess exits | |||
if pipe.poll() is not None: | |||
sys.exit(pipe.returncode) | |||
readable, _, _ = select.select([sys.stdin.fileno(), pipe.stdout.fileno(), pipe.stderr.fileno()], [], []) | |||
if pipe.stdout.fileno() in readable: | |||
self._read_stream_to(pipe.stdout, sys.stdout, stdout_middleware) | |||
if pipe.stderr.fileno() in readable: | |||
self._read_stream_to(pipe.stderr, sys.stderr, stderr_middleware) | |||
if sys.stdin.fileno() in readable: | |||
self._write_stream_to(sys.stdin.buffer, pipe.stdin) | |||
def _read_stream_to(self, input, output, middleware_stack=None): | |||
""" Repeats all bytes written to file-object input to file-object output. | |||
""" | |||
text = input.peek() # preview available bytes | |||
input.read(len(text)) # hacky seek | |||
if text: | |||
text = self._apply_middleware(middleware_stack, text.decode("utf-8")) | |||
output.write(text) | |||
output.flush() | |||
def _write_stream_to(self, input, output): | |||
""" Repeats all bytes read from file-object input to file-object output. | |||
""" | |||
text = input.peek() # preview available bytes | |||
input.read(len(text)) # hacky seek | |||
if isinstance(text, bytes): | |||
output.write(text) | |||
else: | |||
output.write(text.encode("utf-8")) | |||
output.flush() | |||
def _apply_middleware(self, middleware_stack, text): | |||
""" Calls apply(text) method on each `Middleware`, which returns a new text object. | |||
""" | |||
if not middleware_stack: | |||
return text | |||
for middleware in middleware_stack: | |||
text = middleware.apply(text) | |||
return text | |||
if __name__ == '__main__': | |||
wrapper = WrapSubprocess() | |||
cmds = ["rails", "test"] + sys.argv | |||
cmds.remove(__file__) | |||
pipe = subprocess.Popen( | |||
cmds, | |||
stdin=subprocess.PIPE, | |||
stdout=subprocess.PIPE, | |||
stderr=subprocess.PIPE, | |||
env=os.environ | |||
) | |||
wrapper.wrap( | |||
pipe, | |||
stdout_middleware=middleware_stack, | |||
stderr_middleware=middleware_stack, | |||
) | |||
</source> | |||
</blockquote><!-- wrap/modify stdin/stdout/stderr in subprocess --> | |||
</blockquote><!-- SubProcesses --> | </blockquote><!-- SubProcesses --> | ||
Revision as of 11:38, 19 June 2023
If you want to run a command on the host operating system, you'll want to use python's subprocess module. It provides a means of safely spawning a subprocess, and passing parameters in a way that does not expose security vulnerabilities.
There are a lot of variations on how this can be used, these are only the methods I use most frequently.
SubProcesses
Basics
Popen Shortcuts
import subprocess # shortcuts subprocess.check_call(['netstat', '-an'], universal_newlines=True)The Popen objects are the most powerful, but most verbose way of managing processes.
# redirecting streams between processes pipe1 = subprocess.Popen(['tasklist'], stdout=subprocess.PIPE) pipe2 = subprocess.Popen(['find', '/I', 'python'], stdin=pipe1.stdout, stdout=subprocess.PIPE) (stdout, stderr) = pipe2.communicate()Windows
Windows executables are either built as console-applications (require console) or gui-applications (do not show console).
On windows, you can usepythonw.exe
to launch an application without a console.
Regardless of this decision, you can control whether or not a subprocess should open a new console while it runs.import subprocess # don't open new console window for subprocess startupinfo = subprocess.STARTUPINFO() startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW subprocess.Popen(['foo.exe'], startupinfo=startupinfo)wrap stdin/stdout/stderr in subprocess
#!/usr/bin/env python import enum import os import re import select import subprocess import sys class MiddlewareBase: """ Base Class for middleware """ def apply(self, text): raise NotImplementedError() class WrapSubprocess: def wrap(self, pipe, stdout_middleware=None, stderr_middleware=None): """ Wraps a subprocess.Popen(), allowing you to modify it's stdin/stdout/stderr with lists of 'middleware' functions. """ while True: # exit when subprocess exits if pipe.poll() is not None: sys.exit(pipe.returncode) readable, _, _ = select.select([sys.stdin.fileno(), pipe.stdout.fileno(), pipe.stderr.fileno()], [], []) if pipe.stdout.fileno() in readable: self._read_stream_to(pipe.stdout, sys.stdout, stdout_middleware) if pipe.stderr.fileno() in readable: self._read_stream_to(pipe.stderr, sys.stderr, stderr_middleware) if sys.stdin.fileno() in readable: self._write_stream_to(sys.stdin.buffer, pipe.stdin) def _read_stream_to(self, input, output, middleware_stack=None): """ Repeats all bytes written to file-object input to file-object output. """ text = input.peek() # preview available bytes input.read(len(text)) # hacky seek if text: text = self._apply_middleware(middleware_stack, text.decode("utf-8")) output.write(text) output.flush() def _write_stream_to(self, input, output): """ Repeats all bytes read from file-object input to file-object output. """ text = input.peek() # preview available bytes input.read(len(text)) # hacky seek if isinstance(text, bytes): output.write(text) else: output.write(text.encode("utf-8")) output.flush() def _apply_middleware(self, middleware_stack, text): """ Calls apply(text) method on each `Middleware`, which returns a new text object. """ if not middleware_stack: return text for middleware in middleware_stack: text = middleware.apply(text) return text if __name__ == '__main__': wrapper = WrapSubprocess() cmds = ["rails", "test"] + sys.argv cmds.remove(__file__) pipe = subprocess.Popen( cmds, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=os.environ ) wrapper.wrap( pipe, stdout_middleware=middleware_stack, stderr_middleware=middleware_stack, )
Process Exitcodes
import sys sys.exit(1) # exit with fail
Environment Variables
import os os.environ['FOO'] = 'bar'