hook_wrapper.py

"""
Hooks Wrapper Class
"""
import logging
import importlib
from pathlib import Path
from typing import Optional, Dict, cast, List

from .hook_config import HookPackageConfig
from .exceptions import (
    InvalidHookWrapperException,
    HookPackageExecuteFunctionalityException,
)


LOG = logging.getLogger(__name__)
INTERNAL_PACKAGES_ROOT = Path(__file__).parent / ".." / ".." / "hook_packages"


class IacHookWrapper:
    """IacHookWrapper

    An IacHookWrapper instance, upon instantiation, looks up the hook package with the specified hook package ID.
    It provides the "prepare" method, which generates an IaC metadata and output the location of the metadata file.

    Example:
    ```
    hook = IacHookWrapper("terraform")
    metadata_loc = hook.prepare("path/to/iac_project", "path/to/output", True)
    ```
    """

    _hook_name: str
    _config: Optional[HookPackageConfig]

    def __init__(self, hook_name: str):
        """
        Parameters
        ----------
        hook_name: str
            Hook name
        """
        self._hook_name = hook_name
        self._config = None
        self._load_hook_package(hook_name)

    def prepare(
        self,
        output_dir_path: str,
        iac_project_path: Optional[str] = None,
        debug: bool = False,
        aws_profile: Optional[str] = None,
        aws_region: Optional[str] = None,
        skip_prepare_infra: bool = False,
    ) -> str:
        """
        Run the prepare hook to generate the IaC Metadata file.

        Parameters
        ----------
        output_dir_path: str
            the path where the hook can create the generated Metadata files. Required
        iac_project_path: str
            the path where the hook can find the TF application. Default value in current work directory.
        debug: bool
            True/False flag to tell the hooks if should print debugging logs or not. Default is False.
        aws_profile: str
            AWS profile to use. Default is None (use default profile)
        aws_region: str
            AWS region to use. Default is None (use default region)
        skip_prepare_infra: bool
            Flag to skip skip prepare hook if we already have the metadata file. Default is False.

        Returns
        -------
        str
            Path to the generated IaC Metadata file
        """
        LOG.info('Executing prepare hook of hook "%s"', self._hook_name)
        params = {
            "IACProjectPath": iac_project_path if iac_project_path else str(Path.cwd()),
            "OutputDirPath": output_dir_path,
            "Debug": debug,
            "SkipPrepareInfra": skip_prepare_infra,
        }
        if aws_profile:
            params["Profile"] = aws_profile
        if aws_region:
            params["Region"] = aws_region

        output = self._execute("prepare", params)

        metadata_file_loc = None
        iac_applications: Dict[str, Dict] = output.get("iac_applications", {})
        if iac_applications and len(iac_applications) == 1:
            # NOTE: we assume there is only one application in the `iac_applications` dictionary,
            # which is the only case we support right now
            main_application = list(iac_applications.values())[0]
            metadata_file_loc = main_application.get("metadata_file")

        if not metadata_file_loc:
            raise InvalidHookWrapperException("Metadata file path not found in the prepare hook output")

        LOG.debug("Metadata file location - %s", metadata_file_loc)
        return cast(str, metadata_file_loc)

    def _load_hook_package(self, hook_name: str) -> None:
        """Find and load hook package config with given hook name

        Parameters
        ----------
        hook_name: str
            Hook name
        """
        # locate hook package from internal first
        LOG.debug("Looking for internal hook package")
        for child in INTERNAL_PACKAGES_ROOT.iterdir():
            if child.name == hook_name:
                LOG.debug('Loaded internal hook package "%s"', hook_name)
                self._config = HookPackageConfig(child)
                return

        raise InvalidHookWrapperException(f'Cannot locate hook package with hook_name "{hook_name}"')

    def _execute(self, functionality_key: str, params: Optional[Dict] = None) -> Dict:
        """
        Execute a functionality with given key

        Parameters
        ----------
        functionality_key: str
            The key of the functionality
        params: Dict
            A dict of parameters to pass into the execution

        Returns
        -------
        Dict
            the output from the execution
        """
        if not self._config:
            raise InvalidHookWrapperException("Config is missing. You must instantiate a hook with a valid config")

        if functionality_key not in self._config.functionalities:  # pylint: disable=unsupported-membership-test
            raise HookPackageExecuteFunctionalityException(
                f'Functionality "{functionality_key}" is not defined in the hook package'
            )

        functionality = self._config.functionalities[functionality_key]
        if functionality.entry_method:
            return _execute_as_module(functionality.module, functionality.method, params)

        raise InvalidHookWrapperException(f'Functionality "{functionality_key}" is missing an "entry_method"')


def _execute_as_module(module: str, method: str, params: Optional[Dict] = None) -> Dict:
    """
    Execute a module/method with given module and given method

    Parameters
    ----------
    module: str
        the module where the method lives in
    method: str
        the name of the method to execute
    params: Dict
        A dict of parameters to pass into the execution

    Returns
    -------
    Dict
        the output from the execution
    """
    try:
        mod = importlib.import_module(module)
    except ImportError as e:
        raise InvalidHookWrapperException(f'Import error - HookFunctionality module "{module}"') from e

    if not hasattr(mod, method):
        raise InvalidHookWrapperException(f'HookFunctionality module "{module}" has no method "{method}"')

    result = getattr(mod, method)(params)
    return cast(Dict, result)


def get_available_hook_packages_ids() -> List[str]:
    """
    return a list of available hook names.

    Returns
    -------
    List
        The available hook names.
    """
    LOG.debug("Return available internal hook packages")
    hook_packages_ids = []
    for child in INTERNAL_PACKAGES_ROOT.iterdir():
        if child.is_dir() and child.name[0].isalpha():
            hook_packages_ids.append(child.name)

    return hook_packages_ids