CloudBender/cloudbender/hooks.py

import os
import sys
import subprocess
import tempfile
import shutil
from functools import wraps

from .exceptions import InvalidHook

import logging

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


def pulumi_ws(func):
    @wraps(func)
    def decorated(self, *args, **kwargs):
        # setup temp workspace
        self.work_dir = tempfile.mkdtemp(
            dir=tempfile.gettempdir(), prefix="cloudbender-"
        )

        response = func(self, *args, **kwargs)

        # Cleanup temp workspace
        if os.path.exists(self.work_dir):
            shutil.rmtree(self.work_dir)

        return response

    return decorated


# 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:
        raise InvalidHook("Invalid argument {}".format(arguments))