CloudBender/cloudbender/hooks.py

48 lines
1.2 KiB
Python
Raw Normal View History

import sys
import subprocess
from functools import wraps
from .exceptions import InvalidHook
import logging
2022-02-22 10:04:29 +00:00
logger = logging.getLogger(__name__)
def execute_hooks(hooks, stack):
for hook in hooks:
tokens = hook.split()
if tokens[0] in dir(sys.modules[__name__]):
logger.info("Executing hook: {}".format(hook))
globals()[tokens[0]](arguments=tokens[1:], stack=stack)
else:
logger.warning("Unknown hook: {}".format(hook))
def exec_hooks(func):
@wraps(func)
def decorated(self, *args, **kwargs):
execute_hooks(self.hooks.get("pre_" + func.__name__, []), self)
response = func(self, *args, **kwargs)
# Only execute post hook for successful actions
if response == "COMPLETE":
execute_hooks(self.hooks.get("post_" + func.__name__, []), self)
return response
return decorated
2021-09-20 14:19:14 +00:00
# Various hooks
def cmd(stack, arguments):
"""
Generic command via subprocess
"""
try:
hook = subprocess.run(arguments, stdout=subprocess.PIPE)
logger.info(hook.stdout.decode("utf-8"))
except TypeError:
2022-02-22 10:04:29 +00:00
raise InvalidHook("Invalid argument {}".format(arguments))