Source code for runway.cfngin.cfngin

"""CFNgin entrypoint."""
from __future__ import annotations

import logging
import os
import sys
from pathlib import Path
from typing import TYPE_CHECKING, Any, Dict, List, Optional, cast

from yaml.constructor import ConstructorError

from .._logging import PrefixAdaptor
from ..compat import cached_property
from ..config import CfnginConfig
from ..context import CfnginContext
from ..utils import MutableMap, SafeHaven
from .actions import deploy, destroy, diff
from .environment import parse_environment
from .providers.aws.default import ProviderBuilder

if TYPE_CHECKING:
    from .._logging import RunwayLogger
    from ..context import RunwayContext

# explicitly name logger so its not redundant
LOGGER = cast("RunwayLogger", logging.getLogger("runway.cfngin"))


[docs]class CFNgin: """Control CFNgin. Attributes: concurrency: Max number of CFNgin stacks that can be deployed concurrently. If the value is ``0``, will be constrained based on the underlying graph. interactive: Whether or not to prompt the user before taking action. parameters: Combination of the parameters provided when initalizing the class and any environment files that are found. recreate_failed: Destroy and re-create stacks that are stuck in a failed state from an initial deployment when updating. region: The AWS region where CFNgin is currently being executed. sys_path: Working directory. tail: Whether or not to display all CloudFormation events in the terminal. """ concurrency: int interactive: bool parameters: MutableMap recreate_failed: bool region: str sys_path: Path tail: bool def __init__( self, ctx: RunwayContext, parameters: Optional[Dict[str, Any]] = None, sys_path: Path = Path.cwd(), ) -> None: """Instantiate class. Args: ctx: Runway context object. parameters: Parameters from Runway. sys_path: Working directory. """ self.__ctx = ctx self._env_file_name = None self.concurrency = ctx.env.max_concurrent_cfngin_stacks self.interactive = ctx.is_interactive self.parameters = MutableMap() self.recreate_failed = ctx.is_noninteractive self.region = ctx.env.aws_region self.sys_path = sys_path self.tail = bool(ctx.env.debug or ctx.env.verbose) self.parameters.update(self.env_file) if parameters: LOGGER.debug("adding Runway parameters to CFNgin parameters") self.parameters.update(parameters) self._inject_common_parameters() @cached_property def env_file(self) -> MutableMap: """Contents of a CFNgin environment file.""" result: Dict[str, Any] = {} supported_names = [ "{}.env".format(self.__ctx.env.name), "{}-{}.env".format(self.__ctx.env.name, self.region), ] for _, file_name in enumerate(supported_names): file_path = os.path.join(self.sys_path, file_name) if os.path.isfile(file_path): LOGGER.info("found environment file: %s", file_path) self._env_file_name = file_path with open(file_path, "r") as file_: result.update(parse_environment(file_.read())) return MutableMap(**result)
[docs] def deploy(self, force: bool = False, sys_path: Optional[Path] = None) -> None: """Run the CFNgin deploy action. Args: force: Explicitly enable the action even if an environment file is not found. sys_path: Explicitly define a path to work in. If not provided, ``self.sys_path`` is used. """ if self.should_skip(force): return if not sys_path: sys_path = self.sys_path config_file_paths = self.find_config_files(sys_path=sys_path) with SafeHaven( environ=self.__ctx.env.vars, sys_modules_exclude=["awacs", "troposphere"] ): for config_path in config_file_paths: logger = PrefixAdaptor(os.path.basename(config_path), LOGGER) logger.notice("deploy (in progress)") with SafeHaven(sys_modules_exclude=["awacs", "troposphere"]): ctx = self.load(config_path) action = deploy.Action( context=ctx, provider_builder=self._get_provider_builder( ctx.config.service_role ), ) action.execute(concurrency=self.concurrency, tail=self.tail) logger.success("deploy (complete)")
[docs] def destroy(self, force: bool = False, sys_path: Optional[Path] = None) -> None: """Run the CFNgin destroy action. Args: force: Explicitly enable the action even if an environment file is not found. sys_path: Explicitly define a path to work in. If not provided, ``self.sys_path`` is used. """ if self.should_skip(force): return if not sys_path: sys_path = self.sys_path config_file_paths = self.find_config_files(sys_path=sys_path) # destroy should run in reverse to handle dependencies config_file_paths.reverse() with SafeHaven(environ=self.__ctx.env.vars): for config_path in config_file_paths: logger = PrefixAdaptor(config_path.name, LOGGER) logger.notice("destroy (in progress)") with SafeHaven(): ctx = self.load(config_path) action = destroy.Action( context=ctx, provider_builder=self._get_provider_builder( ctx.config.service_role ), ) action.execute( concurrency=self.concurrency, force=True, tail=self.tail ) logger.success("destroy (complete)")
[docs] def load(self, config_path: Path) -> CfnginContext: """Load a CFNgin config into a context object. Args: config_path: Valid path to a CFNgin config file. """ LOGGER.debug("loading CFNgin config: %s", config_path.name) try: config = self._get_config(config_path) config.load() return self._get_context(config, config_path) except ConstructorError as err: if err.problem.startswith( "could not determine a constructor " "for the tag '!" ): LOGGER.error( '"%s" is located in the module\'s root directory ' "and appears to be a CloudFormation template; " "please move CloudFormation templates to a subdirectory", config_path, ) sys.exit(1) raise
[docs] def plan(self, force: bool = False, sys_path: Optional[Path] = None): """Run the CFNgin plan action. Args: force: Explicitly enable the action even if an environment file is not found. sys_path: Explicitly define a path to work in. If not provided, ``self.sys_path`` is used. """ if self.should_skip(force): return if not sys_path: sys_path = self.sys_path config_file_paths = self.find_config_files(sys_path=sys_path) with SafeHaven(environ=self.__ctx.env.vars): for config_path in config_file_paths: logger = PrefixAdaptor(config_path.name, LOGGER) logger.notice("plan (in progress)") with SafeHaven(): ctx = self.load(config_path) action = diff.Action( context=ctx, provider_builder=self._get_provider_builder( ctx.config.service_role ), ) action.execute() logger.success("plan (complete)")
[docs] def should_skip(self, force: bool = False) -> bool: """Determine if action should be taken or not. Args: force: If ``True``, will always return ``False`` meaning the action should not be skipped. """ if force or self.env_file: return False LOGGER.info("skipped; no parameters and environment file not found") return True
def _get_config(self, file_path: Path) -> CfnginConfig: """Initialize a CFNgin config object from a file. Args: file_path: Path to the config file to load. validate: Validate the loaded config. """ return CfnginConfig.parse_file(file_path=file_path, parameters=self.parameters) def _get_context(self, config: CfnginConfig, config_path: Path) -> CfnginContext: """Initialize a CFNgin context object. Args: config: CFNgin config object. config_path: Path to the config file that was provided. """ return CfnginContext( config_path=config_path, config=config, deploy_environment=self.__ctx.env.copy(), force_stacks=[], # placeholder parameters=self.parameters, stack_names=[], # placeholder ) def _get_provider_builder( self, service_role: Optional[str] = None ) -> ProviderBuilder: """Initialize provider builder. Args: service_role: CloudFormation service role. """ if self.interactive: LOGGER.verbose("using interactive AWS provider mode") else: LOGGER.verbose("using default AWS provider mode") return ProviderBuilder( interactive=self.interactive, recreate_failed=self.recreate_failed, region=self.region, service_role=service_role, ) def _inject_common_parameters(self) -> None: """Add common parameters if they don't already exist. Adding these commonly used parameters will remove the need to add lookup support (mainly for environment variable lookups) in places such as ``cfngin_bucket``. Injected Parameters ~~~~~~~~~~~~~~~~~~~ **environment (str)** Taken from the ``DEPLOY_ENVIRONMENT`` environment variable. This will the be current Runway environment being processed. **region (str)** Taken from the ``AWS_REGION`` environment variable. This will be the current region being deployed to. """ if not self.parameters.get("environment"): self.parameters["environment"] = self.__ctx.env.name if not self.parameters.get("region"): self.parameters["region"] = self.region
[docs] @classmethod def find_config_files( cls, exclude: Optional[List[str]] = None, sys_path: Optional[Path] = None ) -> List[Path]: """Find CFNgin config files. Args: exclude: List of file names to exclude. This list is appended to the global exclude list. sys_path: Explicitly define a path to search for config files. Returns: Paths to config files that were found. """ return CfnginConfig.find_config_file(sys_path, exclude=exclude)