-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathcommand.py
More file actions
69 lines (55 loc) · 2.34 KB
/
Copy pathcommand.py
File metadata and controls
69 lines (55 loc) · 2.34 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
import shlex
import subprocess
from .exceptions import CommandExecutionError, CommandParameterError
class Command(object):
process = None
command = 'true'
ignore_output = True
fail_silently = False
required_parameters = None
stdout = subprocess.PIPE
stderr = subprocess.PIPE
def __init__(self, **kwargs):
self.parameters = kwargs
if not self.validate_parameters():
raise CommandParameterError(
'Parameter(s) missing, required parameters: {0}'.format(
', '.join(self.required_parameters)
)
)
def execute(self, ignore_output=None, fail_silently=None, stdin=None, **kwargs):
command = self.get_command()
ignore_output = ignore_output if ignore_output is not None else self.ignore_output
fail_silently = fail_silently if fail_silently is not None else self.fail_silently
# Don't automatically merge with os.environ for security reasons.
# Make this forwarding explicit rather than implicit.
environ = kwargs.pop('environ', None)
shell = kwargs.pop('shell', False)
try:
self.process = subprocess.Popen(
command,
shell=shell,
universal_newlines=True,
env=environ,
stdout=kwargs['stdout'] if 'stdout' in kwargs else self.stdout,
stderr=kwargs['stderr'] if 'stderr' in kwargs else self.stderr,
stdin=subprocess.PIPE,
)
stdout, stderr = self.process.communicate(input=stdin)
except OSError as exc:
raise CommandExecutionError(1, str(exc), self)
if not fail_silently and (stderr or self.process.returncode != 0):
raise CommandExecutionError(self.process.returncode, stderr or '', self)
return True if ignore_output else self.handle_output(stdout)
def validate_parameters(self):
return all(k in self.parameters for k in self.required_parameters or [])
def get_parameters(self):
return self.parameters
def get_command(self):
command = self.command.format(**self.get_parameters())
return shlex.split(str(command))
def handle_output(self, output):
return output
@property
def pid(self):
return self.process.pid if self.process else None