tools.py
4.17 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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
# python standard library
import asyncio
import logging
from os import path
import subprocess
from typing import Any, List
# third party libraries
import yaml
# setup logger for this module
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# load data from yaml file
# ---------------------------------------------------------------------------
def load_yaml(filename: str, default: Any = None) -> Any:
filename = path.expanduser(filename)
try:
f = open(filename, 'r', encoding='utf-8')
except FileNotFoundError:
logger.error(f'Cannot open "{filename}": not found')
except PermissionError:
logger.error(f'Cannot open "{filename}": no permission')
except OSError:
logger.error(f'Cannot open file "{filename}"')
else:
with f:
try:
default = yaml.safe_load(f)
except yaml.YAMLError as e:
if hasattr(e, 'problem_mark'):
mark = e.problem_mark
logger.error(f'File "{filename}" near line {mark.line}, '
f'column {mark.column+1}')
else:
logger.error(f'File "{filename}"')
finally:
return default
# ---------------------------------------------------------------------------
# Runs a script and returns its stdout parsed as yaml, or None on error.
# The script is run in another process but this function blocks waiting
# for its termination.
# ---------------------------------------------------------------------------
def run_script(script: str,
args: List[str] = [],
stdin: str = '',
timeout: int = 2) -> Any:
script = path.expanduser(script)
try:
cmd = [script] + [str(a) for a in args]
p = subprocess.run(cmd,
input=stdin,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=True,
timeout=timeout,
)
except FileNotFoundError:
logger.error(f'Can not execute script "{script}": not found.')
except PermissionError:
logger.error(f'Can not execute script "{script}": wrong permissions.')
except OSError:
logger.error(f'Can not execute script "{script}": unknown reason.')
except subprocess.TimeoutExpired:
logger.error(f'Timeout {timeout}s exceeded while running "{script}".')
except Exception:
logger.error(f'An Exception ocurred running {script}.')
else:
if p.returncode != 0:
logger.error(f'Return code {p.returncode} running "{script}".')
else:
try:
output = yaml.safe_load(p.stdout)
except Exception:
logger.error(f'Error parsing yaml output of "{script}"')
else:
return output
# ----------------------------------------------------------------------------
# Same as above, but asynchronous
# ----------------------------------------------------------------------------
async def run_script_async(script: str,
args: List[str] = [],
stdin: str = '',
timeout: int = 2) -> Any:
script = path.expanduser(script)
args = [str(a) for a in args]
p = await asyncio.create_subprocess_exec(
script, *args,
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.DEVNULL,
)
try:
stdout, stderr = await asyncio.wait_for(
p.communicate(input=stdin.encode('utf-8')),
timeout=timeout
)
except asyncio.TimeoutError:
logger.warning(f'Timeout {timeout}s running script "{script}".')
return
if p.returncode != 0:
logger.error(f'Return code {p.returncode} running "{script}".')
else:
try:
output = yaml.safe_load(stdout.decode('utf-8', 'ignore'))
except Exception:
logger.error(f'Error parsing yaml output of "{script}"')
else:
return output