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
|
"""
DMM Recipe runner
"""
# Imports
import importlib
import sys
import yaml
from loguru import logger
# Initialize logger
logger.add(sys.stderr, format="{time} {level} {message}",
filter="my_module", level="INFO")
logger.add("out.log", backtrace=True, diagnose=True)
@logger.catch
def read_recipe(recipe):
"""
Read the list of stuff that we should be doing.
"""
logger.info(f"Reading recipe: {recipe}")
configfile = open(recipe, "r")
global config
config = (yaml.safe_load(configfile))
logger.debug("Config:\n" + str(config))
sys.path.append(config["module_path"])
logger.debug("Add-on module path(s): ", config['module_path'])
return list(config['recipe'])
@logger.catch
def perform_recipe(recipe):
"""
Runs through the tasks in our recipe.
"""
for task in read_recipe(recipe):
run_task(task)
logger.info("Recipe: %s has completed." % recipe)
@logger.catch
def run_task(task):
"""
Run through a task in a recipe.
We load a module as needed and then unload it when we're done
to conserve memory for installer environments.
"""
logger.info(f"Running task: {task}")
module_name = config['recipe'][task]["module"]
task_config = config['recipe'][task]
try:
module = importlib.import_module(f"dmm.{module_name}")
except ModuleNotFoundError:
logger.error(f"Module: {module_name} could not be found")
try:
function = getattr(module, config["recipe"][task]["function"])
task_params = {key: value for key, value in task_config.items()
if key not in ["module", "function"]}
ecode, result = function(**task_params)
logger.debug(f"Error code : {ecode}, result : {result}")
except AttributeError:
logger.error(f"Module: {module_name} doesn't seem to contain the "
f"function: {config['recipe'][task]["function"]}")
logger.debug("DMM Recipe reader loaded")
|
