Hooks

A hook is a python function, class, or class method that is executed before or after an action is taken for the entire config.

Only the following actions allow pre/post hooks:

deploy

using fields pre_deploy and post_deploy

destroy

using fields pre_destroy and post_destroy

class cfngin.hook

When defining a hook in one of the supported fields, the follow fields can be used.

Lookup Support

The following fields support lookups:

args: Optional[Dict[str, Any]] = {}

A dictionary of arguments to pass to the hook.

This field supports the use of lookups.

Important

Lookups that change the order of execution, like output, can only be used in a post hook but hooks like rxref are able to be used with either pre or post hooks.

Example

pre_deploy:
  - args:
      key: ${val}
data_key: Optional[str] = None

If set, and the hook returns data (a dictionary or pydantic.BaseModel), the results will be stored in CfnginContext.hook_data with the data_key as its key.

Example

pre_deploy:
  - data_key: example-key
enabled: Optional[bool] = True

Whether to execute the hook every CFNgin run. This field provides the ability to execute a hook per environment when combined with a variable.

Example

pre_deploy:
  - enabled: ${enable_example_hook}
path: str

Python importable path to the hook.

Example

pre_deploy:
  - path: runway.cfngin.hooks.command.run_command
required: Optional[bool] = True

Whether to stop execution if the hook fails.

Built-in Hooks

Writing A Custom Hook

A custom hook must be in an executable, importable python package or standalone file. The hook must be importable using your current sys.path. This takes into account the sys_path defined in the config file as well as any paths of package_sources.

When executed, the hook will have various keyword arguments passed to it. The keyword arguments that will always be passed to the hook are context (CfnginContext) and provider (Provider). Anything defined in the args field will also be passed to hook as a keyword argument. For this reason, it is recommended to use an unpack operator (**kwargs) in addition to the keyword arguments the hook requires to ensure future compatibility and account for misconfigurations.

The hook must return True or a truthy object if it was successful. It must return False or a falsy object if it failed. This signifies to CFNgin whether or not to halt execution if the hook is required. If a Dict, MutableMap, or pydantic.BaseModel is returned, it can be accessed by subsequent hooks, lookups, or Blueprints from the context object. It will be stored as context.hook_data[data_key] where data_key is the value set in the hook definition. If data_key is not provided or the type of the returned data is not a Dict, MutableMap, or pydantic.BaseModel, it will not be added to the context object.

Important

When using a pydantic.root_validator() or pydantic.validator() allow_reuse=True must be passed to the decorator. This is because of how hooks are loaded/re-loaded for each usage. Failure to do so will result in an error if the hook is used more than once.

If using boto3 in a hook, use context.get_session() instead of creating a new session to ensure the correct credentials are used.

"""context.get_session() example."""
from __future__ import annotations

from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    from runway.context import CfnginContext


def do_something(context: CfnginContext, **_kwargs: Any) -> None:
    """Do something."""
    s3_client = context.get_session().client("s3")

Example Hook Function

local_path/hooks/my_hook.py
"""My hook."""
from typing import Dict, Optional


def do_something(
    *, is_failure: bool = True, name: str = "Kevin", **_kwargs: str
) -> Optional[Dict[str, str]]:
    """Do something."""
    if is_failure:
        return None
    return {"result": f"You are not a failure {name}."}
local_path/cfngin.yaml
namespace: example
sys_path: ./

pre_deploy:
  - path: hooks.my_hook.do_something
    args:
      is_failure: false

Example Hook Class

Hook classes must implement the interface detailed by the CfnginHookProtocol Protocol. This can be done implicitly or explicitly (by creating a subclass of CfnginHookProtocol).

As shown in this example, HookArgsBaseModel or it’s parent class BaseModel can be used to create self validating and sanitizing data models. These can then be used to parse the values provided in the args field to ensure they match what is expected.

local_path/hooks/my_hook.py
"""My hook."""
import logging
from typing import TYPE_CHECKING, Any, Dict, Optional

from runway.utils import BaseModel
from runway.cfngin.hooks.protocols import CfnginHookProtocol

if TYPE_CHECKING:
    from ...context import CfnginContext

LOGGER = logging.getLogger(__name__)


class MyClassArgs(BaseModel):
    """Arguments for MyClass hook.

    Attributes:
        is_failure: Force the hook to fail if true.
        name: Name used in the response.

    """

    is_failure: bool = False
    name: str


class MyClass(CfnginHookProtocol):
    """My class does a thing.

    Keyword Args:
        is_failure (bool): Force the hook to fail if true.
        name (str): Name used in the response.

    Returns:
        Dict[str, str]: Response message is stored in ``result``.

    Example:
    .. code-block:: yaml

        pre_deploy:
          - path: hooks.my_hook.MyClass
            args:
            is_failure: False
            name: Karen

    """

    args: MyClassArgs

    def __init__(self, context: CfnginContext, **kwargs: Any) -> None:
        """Instantiate class.

        Args:
            context: Context instance. (passed in by CFNgin)
            provider: Provider instance. (passed in by CFNgin)

        """
        kwargs.setdefault("tags", {})

        self.args = self.ARGS_PARSER.parse_obj(kwargs)
        self.args.tags.update(context.tags)
        self.context = context

    def post_deploy(self) -> Optional[Dict[str, str]]:
        """Run during the **post_deploy** stage."""
        if self.args["is_failure"]:
            return None
        return {"result": f"You are not a failure {self.args['name']}."}

    def post_destroy(self) -> None:
        """Run during the **post_destroy** stage."""
        LOGGER.error("post_destroy is not supported by this hook")

    def pre_deploy(self) -> None:
        """Run during the **pre_deploy** stage."""
        LOGGER.error("pre_deploy is not supported by this hook")

    def pre_destroy(self) -> None:
        """Run during the **pre_destroy** stage."""
        LOGGER.error("pre_destroy is not supported by this hook")
local_path/cfngin.yaml
namespace: example
sys_path: ./

pre_deploy:
  - path: hooks.my_hook.MyClass
    args:
      is_failure: False
      name: Karen