Python subprocess: Difference between revisions
Line 47: | Line 47: | ||
== wrap stdin/stdout/stderr in subprocess == | == wrap stdin/stdout/stderr in subprocess == | ||
<blockquote> | <blockquote> | ||
{{ expand | |||
| Wrap a subprocess, mirroring stdin/stdout/stderr from outside to inside | |||
| | |||
It's useful if you want to wrap an interactive process that requires input to stdin. | |||
<source lang="python"> | <source lang="python"> | ||
#!/usr/bin/env python | #!/usr/bin/env python | ||
Line 135: | Line 141: | ||
) | ) | ||
</source> | </source> | ||
}} | |||
</blockquote><!-- wrap/modify stdin/stdout/stderr in subprocess --> | </blockquote><!-- wrap/modify stdin/stdout/stderr in subprocess --> | ||
</blockquote><!-- SubProcesses --> | </blockquote><!-- SubProcesses --> |
Latest revision as of 11:40, 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
Wrap a subprocess, mirroring stdin/stdout/stderr from outside to inside
It's useful if you want to wrap an interactive process that requires input to stdin.#!/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'