Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def start(self):
"""
Start executor with additional checks.
Checks if previous executor isn't running then start process
(executor) and wait until it's started.
"""
if self.pre_start_check():
# Executor or other process is running with same config.
raise AlreadyRunning(self)
Executor.start(self)
self.wait_for(self.after_start_check)
def start(self: ExecutorType) -> ExecutorType:
"""
Start executor with additional checks.
Checks if previous executor isn't running then start process
(executor) and wait until it's started.
:returns: itself
:rtype: Executor
"""
if self.pre_start_check():
# Some other executor (or process) is running with same config:
raise AlreadyRunning(self)
super(Executor, self).start()
self.wait_for(self.check_subprocess)
return self
def start(self):
"""
Reimplements Executor and SimpleExecutor start to allow setting stdin/stdout/stderr/cwd
It may break input/output/communicate, but will ensure child output redirects won't
break parent process by filling the PIPE.
Also, catches ProcessExitedWithError and raise FileNotFoundError if exitcode was 127
"""
if self.pre_start_check():
# Some other executor (or process) is running with same config:
raise AlreadyRunning(self)
if self.process is None:
command = self.command
if not self._shell:
command = self.command_parts
if isinstance(self.stdio, (list, tuple)):
stdin, stdout, stderr = self.stdio
else:
stdin = stdout = stderr = self.stdio
env = os.environ.copy()
env[ENV_UUID] = self._uuid
popen_kwargs = {
"shell": self._shell,
"stdin": stdin,
"stdout": stdout,