Source code for sceptre.config.reader

# -*- coding: utf-8 -*-


This module implements a ConfigReader class, which is responsible for reading
and constructing Stacks.

import collections
import copy
import datetime
import fnmatch
import logging
import sys
import yaml
import json
import tempfile

from os import environ, path, walk
from typing import Set, Tuple
from pathlib import Path
from jinja2 import Environment
from jinja2 import StrictUndefined
from jinja2 import FileSystemLoader
from jinja2 import select_autoescape
from packaging.specifiers import SpecifierSet
from packaging.version import Version

from sceptre import __version__
from sceptre.exceptions import SceptreException
from sceptre.exceptions import DependencyDoesNotExistError
from sceptre.exceptions import InvalidConfigFileError
from sceptre.exceptions import InvalidSceptreDirectoryError
from sceptre.exceptions import VersionIncompatibleError
from sceptre.exceptions import ConfigFileNotFoundError
from sceptre.helpers import sceptreise_path, logging_level
from sceptre.stack import Stack
from sceptre.config import strategies

ConfigAttributes = collections.namedtuple("Attributes", "required optional")

    "dependencies": strategies.list_join,
    "hooks": strategies.child_wins,
    "iam_role": strategies.child_wins,
    "sceptre_role": strategies.child_wins,
    "iam_role_session_duration": strategies.child_wins,
    "sceptre_role_session_duration": strategies.child_wins,
    "notifications": strategies.child_wins,
    "on_failure": strategies.child_wins,
    "parameters": strategies.child_wins,
    "profile": strategies.child_wins,
    "project_code": strategies.child_wins,
    "protect": strategies.child_wins,
    "region": strategies.child_wins,
    "required_version": strategies.child_wins,
    "role_arn": strategies.child_wins,
    "cloudformation_service_role": strategies.child_wins,
    "sceptre_user_data": strategies.child_wins,
    "stack_name": strategies.child_wins,
    "stack_tags": strategies.child_wins,
    "stack_timeout": strategies.child_wins,
    "template_bucket_name": strategies.child_wins,
    "template_key_value": strategies.child_wins,
    "template": strategies.child_wins,
    "template_path": strategies.child_wins,
    "ignore": strategies.child_wins,
    "obsolete": strategies.child_wins,

    {"project_code", "region"},



[docs]class ConfigReader(object): """ Parses YAML configuration files and produces Stack objects. Responsible for loading Resolvers and Hook classes and adding them as constructors to the PyYAML parser. :param context: A SceptreContext. :type sceptre.context.SceptreContext: """ def __init__(self, context): self.logger = logging.getLogger(__name__) self.context = context self.full_config_path = self.context.full_config_path() # Check is valid sceptre project folder self._check_valid_project_path(self.full_config_path) # Add Resolver and Hook classes to PyYAML loader self._add_yaml_constructors(["sceptre.hooks", "sceptre.resolvers"]) if not self.context.user_variables: self.context.user_variables = {} self.templating_vars = {"var": self.context.user_variables} @staticmethod def _iterate_entry_points(group): """ Helper to determine whether to use pkg_resources or importlib.metadata. """ if sys.version_info < (3, 10): from pkg_resources import iter_entry_points return iter_entry_points(group) else: from importlib.metadata import entry_points return entry_points(group=group) def _add_yaml_constructors(self, entry_point_groups): """ Adds PyYAML constructor functions for all classes found registered at the given entry point groups. Classes are registered whereby the node tag is the entry point name. :param entry_point_groups: Names of entry point groups. :type entry_point_groups: list """ self.logger.debug( "Adding yaml constructors for the entry point groups {0}".format( entry_point_groups ) ) def constructor_factory(node_class): """ Returns constructor that will initialise objects from a given node class. :param node_class: Class representing the node. :type node_class: class :returns: Class initialiser. :rtype: func """ # This function signature is required by PyYAML def class_constructor(loader, node): return node_class( loader.construct_object(self.resolve_node_tag(loader, node)) ) # pragma: no cover return class_constructor for group in entry_point_groups: for entry_point in self._iterate_entry_points(group): # Retrieve name and class from entry point node_tag = "!" + node_class = entry_point.load() # Add constructor to PyYAML loader yaml.SafeLoader.add_constructor( node_tag, constructor_factory(node_class) ) self.logger.debug( "Added constructor for %s with node tag %s", str(node_class), node_tag, )
[docs] def resolve_node_tag(self, loader, node): node = copy.copy(node) node.tag = loader.resolve(type(node), node.value, (True, False)) return node
[docs] def construct_stacks(self) -> Tuple[Set[Stack], Set[Stack]]: """ Traverses the files under the command path. For each file encountered, a Stack is constructed using the correct config. Dependencies are traversed and a final set of Stacks is returned. :returns: A set of Stacks. """ stack_map = {} command_stacks = set() root = self.context.full_command_path() if self.context.full_scan: root = self.context.full_config_path() if path.isfile(root): todo = {root} else: todo = set() for directory_name, sub_directories, files in walk(root, followlinks=True): for filename in fnmatch.filter(files, "*.yaml"): if filename.startswith("config."): continue todo.add(path.join(directory_name, filename)) stack_group_configs = {} full_todo = todo.copy() deps_todo = set() while todo: abs_path = todo.pop() rel_path = path.relpath(abs_path, start=self.context.full_config_path()) directory, filename = path.split(rel_path) if directory in stack_group_configs: stack_group_config = stack_group_configs[directory] else: stack_group_config = stack_group_configs[directory] = self._read( path.join(directory, self.context.config_file) ) stack = self._construct_stack(rel_path, stack_group_config) for dep in stack.dependencies: full_dep = str(Path(self.context.full_config_path(), dep)) if not path.exists(full_dep): raise DependencyDoesNotExistError( "{stackname}: Dependency {dep} not found. " "Please make sure that your dependencies stack_outputs " "have their full path from `config` defined.".format(, dep=dep ) ) if full_dep not in full_todo and full_dep not in deps_todo: todo.add(full_dep) deps_todo.add(full_dep) stack_map[sceptreise_path(rel_path)] = stack full_command_path = self.context.full_command_path() if abs_path == full_command_path or abs_path.startswith( full_command_path.rstrip(path.sep) + path.sep ): command_stacks.add(stack) stacks = self.resolve_stacks(stack_map) return stacks, command_stacks
[docs] def resolve_stacks(self, stack_map) -> Set[Stack]: """ Transforms map of Stacks into a set of Stacks, transforms dependencies from a list of Strings (stack names) to a list of Stacks. :param stack_map: Map of stacks, containing dependencies as list of Strings. :type base_config: dict :returns: Set of stacks, containing dependencies as list of Stacks. :rtype: set :raises: sceptre.exceptions.DependencyDoesNotExistError """ stacks = set() for stack in stack_map.values(): if not self.context.ignore_dependencies: for i, dep in enumerate(stack.dependencies): try: if not isinstance(dep, Stack): # If the dependency was inherited from a stack group, it might already # have been mapped and so doesn't need to be mapped again. stack.dependencies[i] = stack_map[sceptreise_path(dep)] except KeyError: raise DependencyDoesNotExistError( "{stackname}: Dependency {dep} not found. " "Valid dependency names are: " "{stackkeys}. " "Please make sure that your dependencies stack_outputs " "have their full path from `config` defined.".format(, dep=dep, stackkeys=", ".join(stack_map.keys()), ) ) # We deduplicate the dependencies using a set here, since it's possible that a given # dependency ends up in the list multiple times. stack.dependencies = list(set(stack.dependencies)) else: stack.dependencies = [] stacks.add(stack) return stacks
def _read(self, rel_path, base_config=None): """ Reads in configuration from one or more YAML files within the Sceptre project folder. :param rel_path: Relative path to config to read. :type rel_path: str :param base_config: Base config to provide defaults. :type base_config: dict :returns: Config read from config files. :rtype: dict """ self.logger.debug("Reading in '%s' files...", rel_path) directory_path, filename = path.split(rel_path) abs_path = path.join(self.full_config_path, rel_path) # Adding properties from class config = { "project_path": self.context.project_path, "stack_group_path": directory_path, } # Adding defaults from base config. if base_config: config.update(base_config) # Check if file exists, but ignore config.yaml as can be inherited. if not path.isfile(abs_path) and not filename.endswith( self.context.config_file ): raise ConfigFileNotFoundError( 'Config file "{0}" not found.'.format(rel_path) ) # Parse and read in the config files. this_config = self._recursive_read(directory_path, filename, config) if "dependencies" in config or "dependencies" in this_config: this_config["dependencies"] = CONFIG_MERGE_STRATEGIES["dependencies"]( this_config.get("dependencies"), config.get("dependencies") ) config.update(this_config) self._check_version(config) self.logger.debug("Config: %s", config) return config def _recursive_read( self, directory_path: str, filename: str, stack_group_config: dict ) -> dict: """ Traverses the directory_path, from top to bottom, reading in all relevant config files. If config attributes are encountered further down the StackGroup they are merged with the parent as defined in the `CONFIG_MERGE_STRATEGIES` dict. :param directory_path: Relative directory path to config to read. :param filename: File name for the config to read. :param stack_group_config: The loaded config file for the StackGroup :returns: Representation of inherited config. """ parent_directory = path.split(directory_path)[0] # Base condition for recursion config = {} if directory_path: config = self._recursive_read( parent_directory, filename, stack_group_config ) # Combine the stack_group_config with the nested config dict config_group = stack_group_config.copy() config_group.update(config) # Read config file and overwrite inherited properties child_config = self._render(directory_path, filename, config_group) or {} for config_key, strategy in CONFIG_MERGE_STRATEGIES.items(): value = strategy(config.get(config_key), child_config.get(config_key)) if value: child_config[config_key] = value config.update(child_config) return config def _write_debug_file(self, content: str, prefix: str) -> str: """ Write some content to a temp file for debug purposes. :param content: the file content to write. :returns: the full path to the temp file. """ with tempfile.NamedTemporaryFile( mode="w", delete=False, prefix=prefix ) as temp_file: temp_file.write(content) temp_file.flush() return def _render(self, directory_path, basename, stack_group_config): """ Reads a configuration file, loads the config file as a template and returns config loaded from the file. :param directory_path: Relative directory path to config to read. :type directory_path: str :param basename: The filename of the config file :type basename: str :param stack_group_config: The loaded config file for the StackGroup :type stack_group_config: dict :returns: rendered template of config file. :rtype: dict """ config = {} abs_directory_path = path.join(self.full_config_path, directory_path) if not path.isfile(path.join(abs_directory_path, basename)): return default_j2_environment_config = { "autoescape": select_autoescape( disabled_extensions=("yaml",), default=True, ), "loader": FileSystemLoader(abs_directory_path), "undefined": StrictUndefined, } j2_environment_config = strategies.dict_merge( default_j2_environment_config, stack_group_config.get("j2_environment", {}), ) j2_environment = Environment(**j2_environment_config) try: template = j2_environment.get_template(basename) except Exception as err: raise SceptreException( f"{Path(directory_path, basename).as_posix()} - {err}" ) from err self.templating_vars.update(stack_group_config) try: rendered_template = template.render( self.templating_vars, command_path=self.context.command_path.split(path.sep), environment_variable=environ, ) except Exception as err: message = f"{Path(directory_path, basename).as_posix()} - {err}" if logging_level() == logging.DEBUG: debug_file_path = self._write_debug_file( json.dumps(self.templating_vars), prefix="vars_" ) message += f"\nTemplating vars saved to: {debug_file_path}" raise SceptreException(message) from err try: config = yaml.safe_load(rendered_template) except Exception as err: message = f"Error parsing {abs_directory_path}{basename}:\n{err}" if logging_level() == logging.DEBUG: debug_file_path = self._write_debug_file( rendered_template, prefix="rendered_" ) message += f"\nRendered template saved to: {debug_file_path}" raise ValueError(message) return config @staticmethod def _check_valid_project_path(config_path): """ Raises an InvalidSceptreDirectoryError if ``path`` is not a directory. :param path: A config directory path. :type path: str :raises: sceptre.exceptions.InvalidSceptreDirectoryError """ if not path.isdir(config_path): raise InvalidSceptreDirectoryError( "Check '{0}' exists.".format(config_path) ) def _check_version(self, config): """ Raises a VersionIncompatibleException when the current Sceptre version does not comply with the configured version requirement. :raises: sceptre.exceptions.VersionIncompatibleException """ sceptre_version = __version__ if "required_version" in config: required_version = config["required_version"] if Version(sceptre_version) not in SpecifierSet(required_version, True): raise VersionIncompatibleError( "Current sceptre version ({0}) does not meet version " "requirements: {1}".format(sceptre_version, required_version) ) @staticmethod def _collect_s3_details(stack_name, config): """ Collects and constructs details for where to store the Template in S3. :param stack_name: Stack name. :type stack_name: str :param config: Config with details. :type config: dict :returns: S3 details. :rtype: dict """ s3_details = None # If the config explicitly sets the template_bucket_name to None, we don't want to enter # this conditional block. if config.get("template_bucket_name") is not None: template_key = "/".join( [ sceptreise_path(stack_name), "{time_stamp}.json".format( time_stamp=datetime.datetime.utcnow().strftime( "%Y-%m-%d-%H-%M-%S-%fZ" ) ), ] ) if "template_key_prefix" in config: prefix = config["template_key_prefix"] template_key = "/".join([prefix.strip("/"), template_key]) s3_details = { "bucket_name": config["template_bucket_name"], "bucket_key": template_key, } return s3_details def _construct_stack(self, rel_path, stack_group_config=None): """ Constructs an individual Stack object from a config path and a base config. :param rel_path: A relative config file path. :type rel_path: str :param stack_group_config: The Stack group config to use as defaults. :type stack_group_config: dict :returns: Stack object :rtype: sceptre.stack.Stack """ directory, filename = path.split(rel_path) if filename == self.context.config_file: pass self.templating_vars["stack_group_config"] = stack_group_config parsed_stack_group_config = self._parsed_stack_group_config(stack_group_config) config = self._read(rel_path, stack_group_config) stack_name = path.splitext(rel_path)[0] # Check for missing mandatory attributes for required_key in REQUIRED_KEYS: if required_key not in config: raise InvalidConfigFileError( "Required attribute '{0}' not found in configuration of '{1}'.".format( required_key, stack_name ) ) s3_details = self._collect_s3_details(stack_name, config) # If disable/enable rollback was specified on the command line, use that. Otherwise, # fall back to the stack config. disable_rollback = self.context.command_params.get("disable_rollback") if disable_rollback is None: disable_rollback = config.get("disable_rollback", False) stack = Stack( name=stack_name, project_code=config["project_code"], template_path=config.get("template_path"), template_handler_config=config.get("template"), region=config["region"], template_bucket_name=config.get("template_bucket_name"), template_key_prefix=config.get("template_key_prefix"), required_version=config.get("required_version"), sceptre_role=config.get("sceptre_role"), iam_role=config.get("iam_role"), sceptre_role_session_duration=config.get("sceptre_role_session_duration"), iam_role_session_duration=config.get("iam_role_session_duration"), profile=config.get("profile"), parameters=config.get("parameters", {}), sceptre_user_data=config.get("sceptre_user_data", {}), hooks=config.get("hooks", {}), s3_details=s3_details, dependencies=config.get("dependencies", []), role_arn=config.get("role_arn"), cloudformation_service_role=config.get("cloudformation_service_role"), protected=config.get("protect", False), tags=config.get("stack_tags", {}), external_name=config.get("stack_name"), notifications=config.get("notifications"), on_failure=config.get("on_failure"), disable_rollback=disable_rollback, stack_timeout=config.get("stack_timeout", 0), ignore=config.get("ignore", False), obsolete=config.get("obsolete", False), stack_group_config=parsed_stack_group_config, config=config, ) del self.templating_vars["stack_group_config"] return stack def _parsed_stack_group_config(self, stack_group_config): """ Remove all config items that are supported by Sceptre and remove the `project_path` and `stack_group_path` added by `read()`. Return a dictionary that has only user-specified config items. """ parsed_config = { key: stack_group_config[key] for key in set(stack_group_config) - set(CONFIG_MERGE_STRATEGIES) } parsed_config.pop("stack_group_path") return parsed_config