waitpid = os.waitpid

waitstatus_to_exitcode = os.waitstatus_to_exitcode

WIFSTOPPED = os.WIFSTOPPED

WSTOPSIG = os.WSTOPSIG

WNOHANG = os.WNOHANG

waitpid = None

waitstatus_to_exitcode = None

WIFSTOPPED = None

WSTOPSIG = None

WNOHANG = None

"""Raised when run() is called with check=True and the process

def __init__(self, returncode, cmd, output=None, stderr=None):

self.returncode = returncode

self.cmd = cmd

self.output = output

self.stderr = stderr

def __str__(self):

if isinstance(self.returncode, int) and self.returncode < 0:

try:

return "Command '%s' died with %r." % (

self.cmd, signal.Signals(-self.returncode))

except ValueError:

return "Command '%s' died with unknown signal %d." % (

self.cmd, -self.returncode)

else:

return (f"Command '{self.cmd}' returned non-zero "

f"exit status {self.returncode}.")

@property

def stdout(self):

"""Alias for output attribute, to match stderr"""

return self.output

@stdout.setter

def stdout(self, value):

# There's no obvious reason to set this, but allow it anyway so

# .stdout is a transparent alias for .output

"""This exception is raised when the timeout expires while waiting for a

def __init__(self, cmd, timeout, output=None, stderr=None):

self.cmd = cmd

self.timeout = timeout

self.output = output

self.stderr = stderr

def __str__(self):

return ("Command '%s' timed out after %s seconds" %

(self.cmd, self.timeout))

@property

def stdout(self):

return self.output

@stdout.setter

def stdout(self, value):

# There's no obvious reason to set this, but allow it anyway so

# .stdout is a transparent alias for .output

def __init__(self, *, dwFlags=0, hStdInput=None, hStdOutput=None,

hStdError=None, wShowWindow=0, lpAttributeList=None):

self.dwFlags = dwFlags

self.hStdInput = hStdInput

self.hStdOutput = hStdOutput

self.hStdError = hStdError

self.wShowWindow = wShowWindow

self.lpAttributeList = lpAttributeList or {"handle_list": []}

def copy(self):

attr_list = self.lpAttributeList.copy()

if 'handle_list' in attr_list:

attr_list['handle_list'] = list(attr_list['handle_list'])

return STARTUPINFO(dwFlags=self.dwFlags,

hStdInput=self.hStdInput,

hStdOutput=self.hStdOutput,

hStdError=self.hStdError,

wShowWindow=self.wShowWindow,

closed = False

def Close(self, CloseHandle=_winapi.CloseHandle):

if not self.closed:

self.closed = True

CloseHandle(self)

def Detach(self):

if not self.closed:

self.closed = True

return int(self)

raise ValueError("already closed")

def __repr__(self):

return "%s(%d)" % (self.__class__.__name__, int(self))

if _active is None:

return

for inst in _active[:]:

res = inst._internal_poll(_deadstate=sys.maxsize)

if res is not None:

try:

_active.remove(inst)

except ValueError:

# This can happen if two threads create a new Popen instance.

# It's harmless that it was already removed, so ignore.

"""Return a list of command-line arguments reproducing the current

args = []

value = sys.flags.optimize

if value > 0:

args.append('-' + 'O' * value)

"""Return a list of command-line arguments reproducing the current

flag_opt_map = {

'debug': 'd',

# 'inspect': 'i',

# 'interactive': 'i',

'dont_write_bytecode': 'B',

'no_site': 'S',

'verbose': 'v',

'bytes_warning': 'b',

'quiet': 'q',

# -O is handled in _optim_args_from_interpreter_flags()

}

args = _optim_args_from_interpreter_flags()

for flag, opt in flag_opt_map.items():

v = getattr(sys.flags, flag)

if v > 0:

args.append('-' + opt * v)

if sys.flags.isolated:

args.append('-I')

else:

if sys.flags.ignore_environment:

args.append('-E')

if sys.flags.no_user_site:

args.append('-s')

if sys.flags.safe_path:

args.append('-P')

# -W options

warnopts = sys.warnoptions[:]

xoptions = getattr(sys, '_xoptions', {})

bytes_warning = sys.flags.bytes_warning

dev_mode = sys.flags.dev_mode

if bytes_warning > 1:

warnopts.remove("error::BytesWarning")

elif bytes_warning:

warnopts.remove("default::BytesWarning")

if dev_mode:

warnopts.remove('default')

for opt in warnopts:

args.append('-W' + opt)

# -X options

if dev_mode:

args.extend(('-X', 'dev'))

for opt in sorted(xoptions):

if opt == 'dev':

# handled above via sys.flags.dev_mode

continue

value = xoptions[opt]

if value is True:

arg = opt

else:

arg = '%s=%s' % (opt, value)

args.extend(('-X', arg))

# Return default text encoding and emit EncodingWarning if

# sys.flags.warn_default_encoding is true.

if sys.flags.warn_default_encoding:

f = sys._getframe()

filename = f.f_code.co_filename

stacklevel = 2

while f := f.f_back:

if f.f_code.co_filename != filename:

break

stacklevel += 1

warnings.warn("'encoding' argument not specified.",

EncodingWarning, stacklevel)

if sys.flags.utf8_mode:

return "utf-8"

else:

"""Run command with arguments. Wait for command to complete or

with Popen(*popenargs, **kwargs) as p:

try:

return p.wait(timeout=timeout)

except: # Including KeyboardInterrupt, wait handled that.

p.kill()

# We don't call p.wait() again as p.__exit__ does that for us.

"""Run command with arguments. Wait for command to complete. If

retcode = call(*popenargs, **kwargs)

if retcode:

cmd = kwargs.get("args")

if cmd is None:

cmd = popenargs[0]

raise CalledProcessError(retcode, cmd)

r"""Run command with arguments and return its output.

for kw in ('stdout', 'check'):

if kw in kwargs:

raise ValueError(f'{kw} argument not allowed, it will be overridden.')

if 'input' in kwargs and kwargs['input'] is None:

# Explicitly passing input=None was previously equivalent to passing an

# empty string. That is maintained here for backwards compatibility.

if kwargs.get('universal_newlines') or kwargs.get('text') or kwargs.get('encoding') \

or kwargs.get('errors'):

empty = ''

else:

empty = b''

kwargs['input'] = empty

return run(*popenargs, stdout=PIPE, timeout=timeout, check=True,

"""A process that has finished running.

def __init__(self, args, returncode, stdout=None, stderr=None):

self.args = args

self.returncode = returncode

self.stdout = stdout

self.stderr = stderr

def __repr__(self):

args = ['args={!r}'.format(self.args),

'returncode={!r}'.format(self.returncode)]

if self.stdout is not None:

args.append('stdout={!r}'.format(self.stdout))

if self.stderr is not None:

args.append('stderr={!r}'.format(self.stderr))

return "{}({})".format(type(self).__name__, ', '.join(args))

__class_getitem__ = classmethod(types.GenericAlias)

def check_returncode(self):

"""Raise CalledProcessError if the exit code is non-zero."""

if self.returncode:

raise CalledProcessError(self.returncode, self.args, self.stdout,

input=None, capture_output=False, timeout=None, check=False, **kwargs):

"""Run command with arguments and return a CompletedProcess instance.

if input is not None:

if kwargs.get('stdin') is not None:

raise ValueError('stdin and input arguments may not both be used.')

kwargs['stdin'] = PIPE

if capture_output:

if kwargs.get('stdout') is not None or kwargs.get('stderr') is not None:

raise ValueError('stdout and stderr arguments may not be used '

'with capture_output.')

kwargs['stdout'] = PIPE

kwargs['stderr'] = PIPE

with Popen(*popenargs, **kwargs) as process:

try:

stdout, stderr = process.communicate(input, timeout=timeout)

except TimeoutExpired as exc:

process.kill()

if _mswindows:

# Windows accumulates the output in a single blocking

# read() call run on child threads, with the timeout

# being done in a join() on those threads. communicate()

# _after_ kill() is required to collect that and add it

# to the exception.

exc.stdout, exc.stderr = process.communicate()

else:

# POSIX _communicate already populated the output so

# far into the TimeoutExpired exception.

process.wait()

raise

except: # Including KeyboardInterrupt, communicate handled that.

process.kill()

# We don't call process.wait() as .__exit__ does that for us.

raise

retcode = process.poll()

if check and retcode:

raise CalledProcessError(retcode, process.args,

output=stdout, stderr=stderr)

"""

# See

# http://msdn.microsoft.com/en-us/library/17w5ykft.aspx

# or search http://msdn.microsoft.com for

# "Parsing C++ Command-Line Arguments"

result = []

needquote = False

for arg in map(os.fsdecode, seq):

bs_buf = []

# Add a space to separate this argument from the others

if result:

result.append(' ')

needquote = (" " in arg) or ("\t" in arg) or not arg

if needquote:

result.append('"')

for c in arg:

if c == '\\':

# Don't know if we need to double yet.

bs_buf.append(c)

elif c == '"':

# Double backslashes.

result.append('\\' * len(bs_buf)*2)

bs_buf = []

result.append('\\"')

else:

# Normal char

if bs_buf:

result.extend(bs_buf)

bs_buf = []

result.append(c)

# Add remaining backslashes, if any.

if bs_buf:

result.extend(bs_buf)

if needquote:

result.extend(bs_buf)

result.append('"')

"""Return (exitcode, output) of executing cmd in a shell.

try:

data = check_output(cmd, shell=True, text=True, stderr=STDOUT,

encoding=encoding, errors=errors)

exitcode = 0

except CalledProcessError as ex:

data = ex.output

exitcode = ex.returncode

if data[-1:] == '\n':

data = data[:-1]

"""Return output (stdout or stderr) of executing cmd in a shell.

"""Check if posix_spawn() can be used for subprocess.

if _mswindows or not hasattr(os, 'posix_spawn'):

# os.posix_spawn() is not available

return False

if ((_env := os.environ.get('_PYTHON_SUBPROCESS_USE_POSIX_SPAWN')) in ('0', '1')):

return bool(int(_env))

if sys.platform in ('darwin', 'sunos5'):

# posix_spawn() is a syscall on both macOS and Solaris,

# and properly reports errors

return True

# Check libc name and runtime libc version

try:

ver = os.confstr('CS_GNU_LIBC_VERSION')

# parse 'glibc 2.28' as ('glibc', (2, 28))

parts = ver.split(maxsplit=1)

if len(parts) != 2:

# reject unknown format

raise ValueError

libc = parts[0]

version = tuple(map(int, parts[1].split('.')))

if sys.platform == 'linux' and libc == 'glibc' and version >= (2, 24):

# glibc 2.24 has a new Linux posix_spawn implementation using vfork

# which properly reports errors to the parent process.

return True

# Note: Don't use the implementation in earlier glibc because it doesn't

# use vfork (even if glibc 2.26 added a pipe to properly report errors

# to the parent process).

except (AttributeError, ValueError, OSError):

# os.confstr() or CS_GNU_LIBC_VERSION value not available

pass

# By default, assume that posix_spawn() does not properly report errors.

""" Execute a child program in a new process.

_child_created = False # Set here since __del__ checks it

def __init__(self, args, bufsize=-1, executable=None,

stdin=None, stdout=None, stderr=None,

preexec_fn=None, close_fds=True,

shell=False, cwd=None, env=None, universal_newlines=None,

startupinfo=None, creationflags=0,

restore_signals=True, start_new_session=False,

pass_fds=(), *, user=None, group=None, extra_groups=None,

encoding=None, errors=None, text=None, umask=-1, pipesize=-1,

process_group=None):

"""Create new Popen instance."""

if not _can_fork_exec:

raise OSError(

errno.ENOTSUP, f"{sys.platform} does not support processes."

)

_cleanup()

# Held while anything is calling waitpid before returncode has been

# updated to prevent clobbering returncode if wait() or poll() are

# called from multiple threads at once. After acquiring the lock,

# code must re-check self.returncode to see if another thread just

# finished a waitpid() call.

self._waitpid_lock = threading.Lock()

self._input = None

self._communication_started = False

if bufsize is None:

bufsize = -1 # Restore default

if not isinstance(bufsize, int):

raise TypeError("bufsize must be an integer")

if stdout is STDOUT:

raise ValueError("STDOUT can only be used for stderr")

if pipesize is None:

pipesize = -1 # Restore default

if not isinstance(pipesize, int):

raise TypeError("pipesize must be an integer")

if _mswindows:

if preexec_fn is not None:

raise ValueError("preexec_fn is not supported on Windows "

"platforms")

else:

# POSIX

if pass_fds and not close_fds:

warnings.warn("pass_fds overriding close_fds.", RuntimeWarning)

close_fds = True

if startupinfo is not None:

raise ValueError("startupinfo is only supported on Windows "

"platforms")

if creationflags != 0:

raise ValueError("creationflags is only supported on Windows "

"platforms")

self.args = args

self.stdin = None

self.stdout = None

self.stderr = None

self.pid = None

self.returncode = None

self.encoding = encoding

self.errors = errors

self.pipesize = pipesize

# Validate the combinations of text and universal_newlines

if (text is not None and universal_newlines is not None

and bool(universal_newlines) != bool(text)):

raise SubprocessError('Cannot disambiguate when both text '

'and universal_newlines are supplied but '

'different. Pass one or the other.')

self.text_mode = encoding or errors or text or universal_newlines

if self.text_mode and encoding is None:

self.encoding = encoding = _text_encoding()

# How long to resume waiting on a child after the first ^C.

# There is no right value for this. The purpose is to be polite

# yet remain good for interactive users trying to exit a tool.

self._sigint_wait_secs = 0.25 # 1/xkcd221.getRandomNumber()

self._closed_child_pipe_fds = False

if self.text_mode:

if bufsize == 1:

line_buffering = True

# Use the default buffer size for the underlying binary streams

# since they don't support line buffering.

bufsize = -1

else:

line_buffering = False

if process_group is None:

process_group = -1 # The internal APIs are int-only

gid = None

if group is not None:

if not hasattr(os, 'setregid'):

raise ValueError("The 'group' parameter is not supported on the "

"current platform")

elif isinstance(group, str):

try:

import grp

except ImportError:

raise ValueError("The group parameter cannot be a string "

"on systems without the grp module")

gid = grp.getgrnam(group).gr_gid

elif isinstance(group, int):

gid = group

else:

raise TypeError("Group must be a string or an integer, not {}"

.format(type(group)))

if gid < 0:

raise ValueError(f"Group ID cannot be negative, got {gid}")

gids = None

if extra_groups is not None:

if not hasattr(os, 'setgroups'):

raise ValueError("The 'extra_groups' parameter is not "

"supported on the current platform")

elif isinstance(extra_groups, str):

raise ValueError("Groups must be a list, not a string")

gids = []

for extra_group in extra_groups:

if isinstance(extra_group, str):

try:

import grp

except ImportError:

raise ValueError("Items in extra_groups cannot be "

"strings on systems without the "

"grp module")

gids.append(grp.getgrnam(extra_group).gr_gid)

elif isinstance(extra_group, int):

gids.append(extra_group)

else:

raise TypeError("Items in extra_groups must be a string "

"or integer, not {}"

.format(type(extra_group)))

# make sure that the gids are all positive here so we can do less

# checking in the C code

for gid_check in gids:

if gid_check < 0:

raise ValueError(f"Group ID cannot be negative, got {gid_check}")

uid = None

if user is not None:

if not hasattr(os, 'setreuid'):

raise ValueError("The 'user' parameter is not supported on "

"the current platform")

elif isinstance(user, str):

try:

import pwd

except ImportError:

raise ValueError("The user parameter cannot be a string "

"on systems without the pwd module")

uid = pwd.getpwnam(user).pw_uid

elif isinstance(user, int):

uid = user

else:

raise TypeError("User must be a string or an integer")

if uid < 0:

raise ValueError(f"User ID cannot be negative, got {uid}")

# Input and output objects. The general principle is like

# this:

#

# Parent Child

# ------ -----

# p2cwrite ---stdin---> p2cread

# c2pread <--stdout--- c2pwrite

# errread <--stderr--- errwrite

#

# On POSIX, the child objects are file descriptors. On

# Windows, these are Windows file handles. The parent objects

# are file descriptors on both platforms. The parent objects

# are -1 when not using PIPEs. The child objects are -1