我们从Python开源项目中，提取了以下20个代码示例，用于说明如何使用subprocess.html()。
def run(self, terms, variables, **kwargs): ret = [] for term in terms: ''' http://docs.python.org/2/library/subprocess.html#popen-constructor The shell argument (which defaults to False) specifies whether to use the shell as the program to execute. If shell is True, it is recommended to pass args as a string rather than as a sequence https://github.com/ansible/ansible/issues/6550 ''' term = str(term) p = subprocess.Popen(term, cwd=self._loader.get_basedir(), shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE) (stdout, stderr) = p.communicate() if p.returncode == 0: ret.append(stdout.decode("utf-8").rstrip()) else: raise AnsibleError("lookup_plugin.pipe(%s) returned %d" % (term, p.returncode)) return ret
def get_ip_address(self, interface): # Not a simple subprocess.call due to the need for a PIPE in the command # https://docs.python.org/2/library/subprocess.html#replacing-shell-pipeline command_line_1 = "ip addr show " + interface args1 = shlex.split(command_line_1) command_line_2 = "grep -Po 'inet \K[\d.]+'" args2 = shlex.split(command_line_2) command_line_1 = subprocess.Popen(args1, stdout=subprocess.PIPE) command_line_2 = subprocess.Popen(args2, stdin=command_line_1.stdout, stdout=subprocess.PIPE) command_line_1.stdout.close() ip_address = command_line_2.communicate()[0] ip_address = ip_address.rstrip() if ip_address == '': ip_address = 'NOT CONNECTED' return str(ip_address)
def __del__(self, **kwargs): """ Destruct object and perform cleanup. If the process is still running, it is stopped and warning is generated. Process should be stopped by calling `self.stop()` or `self.cleanup()` or by exiting the context manager. Warning: "It is not guaranteed that __del__() methods are called for objects that still exist when the interpreter exits."`Data model <https://docs.python.org/3.5/reference/datamodel.html>`_ Warning: Destructor of object is called even in case of unsuccessful initialization. If __init__ raised an exception, some attributes may be uninitialized. Therefore we need to check `self._popen_initialized` before calling methods inherited from popen which access attributes inherited from popen. """ stop_needed = False # process needs to be stopped if self._popen_initialized and self.poll() is None: # process is running stop_needed = True warnings.warn( 'Process {} was not stopped correctly. Stopping it by destructor, which is not always safe!'.format( type(self).__name__), ResourceWarning) super().__del__() self.cleanup(stop=stop_needed)
def create_proc(self,line, env = None, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=None): """ Create subprocess Popen https://docs.python.org/3/library/subprocess.html :param line: line to evaluate :param env: Env variable :param stdin: stdin type :param stdout: stdout type :param stderr: stderr type read more about types https://docs.python.org/3/library/subprocess.html#subprocess.DEVNULL :return: Popen >>> UnixLang().create_proc('echo "DEVNULL|PIPE|STDOUT"', stdout=subprocess.PIPE).communicate()[0].decode().strip() 'DEVNULL|PIPE|STDOUT' """ proc = subprocess.Popen([line], stdin=stdin, stdout=stdout, stderr=stderr, shell=True, env=env, cwd=self.current_dir) return proc
def ExecLinkWrapper(self, arch, use_separate_mspdbsrv, *args): """Filter diagnostic output from link that looks like: ' Creating library ui.dll.lib and object ui.dll.exp' This happens when there are exports from the dll or exe. """ env = self._GetEnv(arch) if use_separate_mspdbsrv == 'True': self._UseSeparateMspdbsrv(env, args) if sys.platform == 'win32': args = list(args) # *args is a tuple by default, which is read-only. args[0] = args[0].replace('/', '\\') # https://docs.python.org/2/library/subprocess.html: # "On Unix with shell=True [...] if args is a sequence, the first item # specifies the command string, and any additional items will be treated as # additional arguments to the shell itself. That is to say, Popen does the # equivalent of: # Popen(['/bin/sh', '-c', args[0], args[1], ...])" # For that reason, since going through the shell doesn't seem necessary on # non-Windows don't do that there. link = subprocess.Popen(args, shell=sys.platform == 'win32', env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) # Read output one line at a time as it shows up to avoid OOM failures when # GBs of output is produced. for line in link.stdout: if (not line.startswith(' Creating library ') and not line.startswith('Generating code') and not line.startswith('Finished generating code')): print line, return link.wait()
def RunCommand(command, msvc_arch=None, env=None, fail_hard=True): """Run command and return success (True) or failure; or if fail_hard is True, exit on failure. If msvc_arch is set, runs the command in a shell with the msvc tools for that architecture.""" if msvc_arch and sys.platform == 'win32': command = GetVSVersion().SetupScript(msvc_arch) + ['&&'] + command # https://docs.python.org/2/library/subprocess.html: # "On Unix with shell=True [...] if args is a sequence, the first item # specifies the command string, and any additional items will be treated as # additional arguments to the shell itself. That is to say, Popen does the # equivalent of: # Popen(['/bin/sh', '-c', args[0], args[1], ...])" # # We want to pass additional arguments to command[0], not to the shell, # so manually join everything into a single string. # Annoyingly, for "svn co url c:\path", pipes.quote() thinks that it should # quote c:\path but svn can't handle quoted paths on Windows. Since on # Windows follow-on args are passed to args[0] instead of the shell, don't # do the single-string transformation there. if sys.platform != 'win32': command = ' '.join([pipes.quote(c) for c in command]) print 'Running', command if subprocess.call(command, env=env, shell=True) == 0: return True print 'Failed.' if fail_hard: sys.exit(1) return False
def run(self, paths = []): p = get_path(paths) rekitRoot = get_rekit_root(p) reportType = '' if is_app_test_folder(p): reportType = 'app' elif is_cli_test_folder(p): reportType = 'cli' reportPath = os.path.join(rekitRoot, 'coverage', reportType, 'lcov-report/index.html') webbrowser.open('file://' + reportPath)
def is_enabled(self, paths = []): if not self.is_visible(paths): return False p = get_path(paths) rekitRoot = get_rekit_root(p) reportType = '' if is_app_test_folder(p): reportType = 'app' elif is_cli_test_folder(p): reportType = 'cli' reportPath = os.path.join(rekitRoot, 'coverage', reportType, 'lcov-report/index.html') return os.path.exists(reportPath)
def run(self, terms, variables=None, **kwargs): ret = [] for term in terms: ''' http://docs.python.org/2/library/subprocess.html#popen-constructor The shell argument (which defaults to False) specifies whether to use the shell as the program to execute. If shell is True, it is recommended to pass args as a string rather than as a sequence https://github.com/ansible/ansible/issues/6550 ''' name, params = _parse_parameters(term) if params['regenerate']: try: generate_password(name, params['length'], params['symbols'], True) display.vvv('Generated password for %s' % name) except Exception as e: raise AnsibleError("lookup_plugin.pass(%s) returned %s" % (term, e.message)) try: password = get_password(term) except: try: generate_password(name, params['length'], params['symbols']) display.vvv('Generated password for %s' % name) password = get_password(name) except Exception as e: raise AnsibleError("lookup_plugin.pass(%s) returned %s" % (term, e.message)) ret.append(password) return ret
def __exit__(self, exc_type, exc_val, exc_tb): """ Stop the process and then do cleanup. :param exc_type: Exception type :param exc_val: Exception value :param exc_tb: Exception traceback information """ self.stop() # "...on exit, standard file descriptors are closed, and the process is waited for." # `subprocess — Subprocess management <https://docs.python.org/3/library/subprocess.html#subprocess.Popen>`_ super().__exit__(exc_type, exc_val, exc_tb) self.update() # close files used for feedback self.cleanup()
def communicate(self, input=None, timeout=None, nag_timer=None, nag_max=None): """Adds timeout and callbacks support. Returns (stdout, stderr) like subprocess.Popen().communicate(). - The process will be killed after |timeout| seconds and returncode set to TIMED_OUT. - If the subprocess runs for |nag_timer| seconds without producing terminal output, print a warning to stderr. """ self.timeout = timeout self.nag_timer = nag_timer self.nag_max = nag_max if (not self.timeout and not self.nag_timer and not self.stdout_cb and not self.stderr_cb): return super(Popen, self).communicate(input) if self.timeout and self.shell: raise TypeError( 'Using timeout and shell simultaneously will cause a process leak ' 'since the shell will be killed instead of the child process.') stdout = None stderr = None # Convert to a lambda to workaround python's deadlock. # http://docs.python.org/library/subprocess.html#subprocess.Popen.wait # When the pipe fills up, it would deadlock this process. if self.stdout and not self.stdout_cb and not self.stdout_is_void: stdout = [] self.stdout_cb = stdout.append if self.stderr and not self.stderr_cb and not self.stderr_is_void: stderr = [] self.stderr_cb = stderr.append self._tee_threads(input) if stdout is not None: stdout = ''.join(stdout) if stderr is not None: stderr = ''.join(stderr) return (stdout, stderr)
def ExecLinkWrapper(self, arch, use_separate_mspdbsrv, *args): """Filter diagnostic output from link that looks like: ' Creating library ui.dll.lib and object ui.dll.exp' This happens when there are exports from the dll or exe. """ env = self._GetEnv(arch) if use_separate_mspdbsrv == 'True': self._UseSeparateMspdbsrv(env, args) if sys.platform == 'win32': args = list(args) # *args is a tuple by default, which is read-only. args[0] = args[0].replace('/', '\\') # https://docs.python.org/2/library/subprocess.html: # "On Unix with shell=True [...] if args is a sequence, the first item # specifies the command string, and any additional items will be treated as # additional arguments to the shell itself. That is to say, Popen does the # equivalent of: # Popen(['/bin/sh', '-c', args[0], args[1], ...])" # For that reason, since going through the shell doesn't seem necessary on # non-Windows don't do that there. link = subprocess.Popen(args, shell=sys.platform == 'win32', env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) out, _ = link.communicate() for line in out.splitlines(): if (not line.startswith(' Creating library ') and not line.startswith('Generating code') and not line.startswith('Finished generating code')): print line return link.returncode
def run(self): si = None if hasattr(subprocess, "STARTUPINFO"): si = subprocess.STARTUPINFO() si.dwFlags |= subprocess.STARTF_USESHOWWINDOW try: #si.wShowWindow = subprocess.SW_HIDE # default envPATH = os.environ['PATH'] + LOCAL_PATH s = sublime.load_settings("Rekit.sublime-settings") if s.get('node_dir') and envPATH.find(s.get('node_dir')) == -1: envPATH = envPATH + os.pathsep + s.get('node_dir') if s.get('npm_dir') and envPATH.find(s.get('npm_dir')) == -1: envPATH = envPATH + os.pathsep + s.get('npm_dir') # https://docs.python.org/2/library/subprocess.html # Note If specified, env must provide any variables required for the program to execute. # On Windows, in order to run a side-by-side assembly the specified env **must** include a valid SystemRoot. envObj = { 'PATH': envPATH, 'SYSTEMROOT': os.environ['SYSTEMROOT'] } p = subprocess.Popen(self.command, cwd=self.working_dir, env=envObj, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=1, startupinfo=si) for line in iter(p.stdout.readline, b''): line2 = line.decode().strip('\r\n') # only show output for mocha if re.search(r'run_test\.js|build\.js', self.command[1]) is not None: show_rekit_output(line2) if self.on_done: self.on_done() except subprocess.CalledProcessError as e: # show_rekit_output(str(e)) show_rekit_output('running node failed:') show_rekit_output(str(e)) sublime.error_message(str(e)) except OSError as e: if e.errno == 2: main_thread(sublime.error_message, "Node binary could not be found in PATH\nConsider using the node_dir and npm_dir settings for the Rekit plugin\n\nPATH is: %s" % os.environ['PATH']) else: show_rekit_output('running node failed:') show_rekit_output(str(e)) raise e except Exception as e: show_rekit_output('running node failed:') show_rekit_output(str(e))
def subprocess_Popen(command, **params): """ Utility function to work around windows behavior that open windows. :see: call_subprocess_Popen and output_subprocess_Popen """ startupinfo = None if os.name == 'nt': startupinfo = subprocess.STARTUPINFO() try: startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW except AttributeError: startupinfo.dwFlags |= subprocess._subprocess.STARTF_USESHOWWINDOW # Anaconda for Windows does not always provide .exe files # in the PATH, they also have .bat files that call the corresponding # executable. For instance, "g++.bat" is in the PATH, not "g++.exe" # Unless "shell=True", "g++.bat" is not executed when trying to # execute "g++" without extensions. # (Executing "g++.bat" explicitly would also work.) params['shell'] = True # "If shell is True, it is recommended to pass args as a string rather than as a sequence." (cite taken from https://docs.python.org/2/library/subprocess.html#frequently-used-arguments) # In case when command arguments have spaces, passing a command as a list will result in incorrect arguments break down, and consequently # in "The filename, directory name, or volume label syntax is incorrect" error message. # Passing the command as a single string solves this problem. if isinstance(command, list): command = ' '.join(command) # Using the dummy file descriptors below is a workaround for a # crash experienced in an unusual Python 2.4.4 Windows environment # with the default None values. stdin = None if "stdin" not in params: stdin = open(os.devnull) params['stdin'] = stdin.fileno() try: proc = subprocess.Popen(command, startupinfo=startupinfo, **params) finally: if stdin is not None: stdin.close() return proc
