"""CFNgin config."""
from __future__ import annotations
import logging
import re
import sys
from pathlib import Path
from string import Template
from typing import (
TYPE_CHECKING,
AbstractSet,
Any,
Dict,
List,
Mapping,
MutableMapping,
Optional,
Union,
cast,
)
import yaml
from ..cfngin import exceptions
from ..cfngin.lookups.registry import register_lookup_handler
from ..cfngin.utils import SourceProcessor
from ..exceptions import ConfigNotFound
from ..utils import merge_dicts
from .components.runway import (
RunwayDeploymentDefinition,
RunwayTestDefinition,
RunwayVariablesDefinition,
)
from .models.cfngin import (
CfnginConfigDefinitionModel,
CfnginHookDefinitionModel,
CfnginPackageSourcesDefinitionModel,
CfnginStackDefinitionModel,
)
from .models.runway import RunwayConfigDefinitionModel, RunwayFutureDefinitionModel
if TYPE_CHECKING:
from packaging.specifiers import SpecifierSet
from pydantic import BaseModel
LOGGER = logging.getLogger(__name__)
[docs]class BaseConfig:
"""Base class for configurations."""
file_path: Path
_data: BaseModel
[docs] def __init__(self, data: BaseModel, *, path: Optional[Path] = None) -> None:
"""Instantiate class.
Args:
data: The data model of the config file.
path: Path to the config file.
"""
self._data = data.copy()
self.file_path = path.resolve() if path else Path.cwd()
[docs] def dump(
self,
*,
by_alias: bool = False,
exclude: Optional[
Union[AbstractSet[Union[int, str]], Mapping[Union[int, str], Any]]
] = None,
exclude_defaults: bool = False,
exclude_none: bool = False,
exclude_unset: bool = True,
include: Optional[
Union[AbstractSet[Union[int, str]], Mapping[Union[int, str], Any]]
] = None,
) -> str:
"""Dump model to a YAML string.
Args:
by_alias: Whether field aliases should be used as keys in the
returned dictionary.
exclude: Fields to exclude from the returned dictionary.
exclude_defaults: Whether fields which are equal to their default
values (whether set or otherwise) should be excluded from
the returned dictionary.
exclude_none: Whether fields which are equal to None should be
excluded from the returned dictionary.
exclude_unset: Whether fields which were not explicitly set when
creating the model should be excluded from the returned
dictionary.
include: Fields to include in the returned dictionary.
"""
return yaml.dump(
self._data.dict(
by_alias=by_alias,
exclude=exclude, # type: ignore
exclude_defaults=exclude_defaults,
exclude_none=exclude_none,
exclude_unset=exclude_unset,
include=include, # type: ignore
),
default_flow_style=False,
)
[docs] @classmethod
def find_config_file(cls, path: Path) -> Optional[Path]:
"""Find a config file in the provided path.
Args:
path: The path to search for a config file.
"""
raise NotImplementedError # cov: ignore
[docs]class CfnginConfig(BaseConfig):
"""Python representation of a CFNgin config file.
This is used internally by CFNgin to parse and validate a YAML formatted
CFNgin configuration file, but can also be used in scripts to generate a
CFNgin config file before handing it off to CFNgin to deploy/destroy.
Example::
from runway.cfngin.config import dump, Config, Stack
vpc = Stack({
"name": "vpc",
"class_path": "blueprints.VPC"})
config = Config()
config.namespace = "prod"
config.stacks = [vpc]
print dump(config)
"""
EXCLUDE_REGEX = r"runway(\..*)?\.(yml|yaml)"
"""Regex for file names to exclude when looking for config files."""
EXCLUDE_LIST = ["bitbucket-pipelines.yml", "buildspec.yml", "docker-compose.yml"]
"""Explicit files names to ignore when looking for config files."""
cfngin_bucket: Optional[str]
"""Bucket to use for CFNgin resources. (e.g. CloudFormation templates).
May be an empty string.
"""
cfngin_bucket_region: Optional[str]
"""Explicit region to use for :attr:`CfnginConfig.cfngin_bucket`"""
cfngin_cache_dir: Path
"""Local directory to use for caching."""
log_formats: Dict[str, str]
"""Custom formatting for log messages."""
lookups: Dict[str, str]
"""Register custom lookups."""
mappings: Dict[str, Dict[str, Dict[str, Any]]]
"""Mappings that will be added to all stacks."""
namespace: str
"""Namespace to prepend to everything."""
namespace_delimiter: str
"""Character used to separate :attr:`CfnginConfig.namespace` and anything it prepends."""
package_sources: CfnginPackageSourcesDefinitionModel
"""Remote source locations."""
persistent_graph_key: Optional[str] = None
"""S3 object key were the persistent graph is stored."""
post_deploy: List[CfnginHookDefinitionModel]
"""Hooks to run after a deploy action."""
post_destroy: List[CfnginHookDefinitionModel]
"""Hooks to run after a destroy action."""
pre_deploy: List[CfnginHookDefinitionModel]
"""Hooks to run before a deploy action."""
pre_destroy: List[CfnginHookDefinitionModel]
"""Hooks to run before a destroy action."""
service_role: Optional[str]
"""IAM role for CloudFormation to use."""
stacks: List[CfnginStackDefinitionModel]
"""Stacks to be processed."""
sys_path: Optional[Path]
"""Relative or absolute path to use as the work directory."""
tags: Optional[Dict[str, str]]
"""Tags to apply to all resources."""
template_indent: int
"""Spaces to use per-indent level when outputting a template to json."""
_data: CfnginConfigDefinitionModel
[docs] def __init__(
self,
data: CfnginConfigDefinitionModel,
*,
path: Optional[Path] = None,
work_dir: Optional[Path] = None,
) -> None:
"""Instantiate class.
Args:
data: The data model of the config file.
path: Path to the config file.
work_dir: Working directory.
"""
super().__init__(data, path=path)
self.cfngin_bucket = self._data.cfngin_bucket
self.cfngin_bucket_region = self._data.cfngin_bucket_region
if self._data.cfngin_cache_dir:
self.cfngin_cache_dir = self._data.cfngin_cache_dir
elif work_dir:
self.cfngin_cache_dir = work_dir / "cache"
elif path:
self.cfngin_cache_dir = path.parent / ".runway" / "cache"
else:
self.cfngin_cache_dir = Path().cwd() / ".runway" / "cache"
self.log_formats = self._data.log_formats
self.lookups = self._data.lookups
self.mappings = self._data.mappings
self.namespace = self._data.namespace
self.namespace_delimiter = self._data.namespace_delimiter
self.package_sources = self._data.package_sources
self.persistent_graph_key = self._data.persistent_graph_key
self.post_deploy = cast(List[CfnginHookDefinitionModel], self._data.post_deploy)
self.post_destroy = cast(
List[CfnginHookDefinitionModel], self._data.post_destroy
)
self.pre_deploy = cast(List[CfnginHookDefinitionModel], self._data.pre_deploy)
self.pre_destroy = cast(List[CfnginHookDefinitionModel], self._data.pre_destroy)
self.service_role = self._data.service_role
self.stacks = cast(List[CfnginStackDefinitionModel], self._data.stacks)
self.sys_path = self._data.sys_path
self.tags = self._data.tags
self.template_indent = self._data.template_indent
[docs] def load(self) -> None:
"""Load config options into the current environment/session."""
if self.sys_path:
LOGGER.debug("appending to sys.path: %s", self.sys_path)
sys.path.append(str(self.sys_path))
LOGGER.debug("sys.path: %s", sys.path)
if self.lookups:
for key, handler in self.lookups.items():
register_lookup_handler(key, handler)
[docs] @classmethod
def find_config_file( # type: ignore pylint: disable=arguments-differ
cls, path: Optional[Path] = None, *, exclude: Optional[List[str]] = None
) -> List[Path]:
"""Find a config file in the provided path.
Args:
path: The path to search for a config file.
exclude: List of file names to exclude. This list is appended to
the global exclude list.
Raises:
ConfigNotFound: Could not find a config file in the provided path.
ValueError: More than one config file found in the provided path.
"""
if not path:
path = Path.cwd()
elif path.is_file():
return [path]
exclude = exclude or []
result: List[Path] = []
exclude.extend(cls.EXCLUDE_LIST)
yml_files = list(path.glob("*.yml"))
yml_files.extend(list(path.glob("*.yaml")))
for f in yml_files:
if (
re.match(cls.EXCLUDE_REGEX, f.name)
or f.name in exclude
or f.name.startswith(".")
):
continue # cov: ignore
result.append(f)
result.sort()
return result
[docs] @classmethod
def parse_file(
cls,
*,
path: Optional[Path] = None,
file_path: Optional[Path] = None,
parameters: Optional[MutableMapping[str, Any]] = None,
work_dir: Optional[Path] = None,
**kwargs: Any,
) -> CfnginConfig:
"""Parse a YAML file to create a config object.
Args:
path: The path to search for a config file.
file_path: Exact path to a file to parse.
parameters: Values to use when resolving a raw config.
work_dir: Explicit working directory.
Raises:
ConfigNotFound: Provided config file was not found.
"""
if file_path:
if not file_path.is_file():
raise ConfigNotFound(path=file_path)
return cls.parse_raw(
file_path.read_text(),
path=file_path,
parameters=parameters or {},
work_dir=work_dir,
**kwargs,
)
if path:
found = cls.find_config_file(path)
if len(found) > 1:
raise ValueError(f"more than one config files found: {found}")
return cls.parse_file(
file_path=found[0],
parameters=parameters or {},
work_dir=work_dir,
**kwargs,
)
raise ValueError("must provide path or file_path")
[docs] @classmethod
def parse_obj(
cls, obj: Any, *, path: Optional[Path] = None, work_dir: Optional[Path] = None
) -> CfnginConfig:
"""Parse a python object.
Args:
obj: A python object to parse as a CFNgin config.
path: The path to the config file that was parsed into the object.
work_dir: Working directory.
"""
return cls(
CfnginConfigDefinitionModel.parse_obj(obj), path=path, work_dir=work_dir
)
[docs] @classmethod
def parse_raw(
cls,
data: str,
*,
parameters: Optional[MutableMapping[str, Any]] = None,
path: Optional[Path] = None,
skip_package_sources: bool = False,
work_dir: Optional[Path] = None,
) -> CfnginConfig:
"""Parse raw data.
Args:
data: The raw data to parse.
parameters: Values to use when resolving a raw config.
path: The path to search for a config file.
skip_package_sources: Skip processing package sources.
work_dir: Explicit working directory.
"""
if not parameters:
parameters = {}
pre_rendered = cls.resolve_raw_data(data, parameters=parameters)
if skip_package_sources:
return cls.parse_obj(yaml.safe_load(pre_rendered))
config_dict = yaml.safe_load(
cls.process_package_sources(
pre_rendered, parameters=parameters, work_dir=work_dir
)
)
return cls.parse_obj(config_dict, path=path)
[docs] @classmethod
def process_package_sources(
cls,
raw_data: str,
*,
parameters: Optional[MutableMapping[str, Any]] = None,
work_dir: Optional[Path] = None,
) -> str:
"""Process the package sources defined in a rendered config.
Args:
raw_data: Raw configuration data.
cache_dir: Directory to use when caching remote sources.
parameters: Values to use when resolving a raw config.
work_dir: Explicit working directory.
"""
config = yaml.safe_load(raw_data) or {}
processor = SourceProcessor(
sources=CfnginPackageSourcesDefinitionModel.parse_obj(
config.get("package_sources", {}) # type: ignore
),
cache_dir=Path(
config.get(
"cfngin_cache_dir", (work_dir or Path().cwd() / ".runway") / "cache"
)
),
)
processor.get_package_sources()
if processor.configs_to_merge:
for i in processor.configs_to_merge:
LOGGER.debug("merging in remote config: %s", i)
with open(i, "rb") as opened_file:
config = merge_dicts(yaml.safe_load(opened_file), config)
return cls.resolve_raw_data(yaml.dump(config), parameters=parameters or {})
return raw_data
[docs] @staticmethod
def resolve_raw_data(
raw_data: str, *, parameters: Optional[MutableMapping[str, Any]] = None
) -> str:
"""Resolve raw data.
Args:
raw_data: Raw configuration data.
parameters: Values to use when resolving a raw config.
Raises:
MissingEnvironment: A value required by the config was not provided
in parameters.
"""
if not parameters:
parameters = {}
template = Template(raw_data)
try:
rendered = template.substitute(**parameters)
except KeyError as err:
raise exceptions.MissingEnvironment(err.args[0]) from None
except ValueError:
rendered = template.safe_substitute(**parameters)
return rendered
[docs]class RunwayConfig(BaseConfig):
"""Python representation of a Runway config file."""
ACCEPTED_NAMES = ["runway.yml", "runway.yaml"]
deployments: List[RunwayDeploymentDefinition]
file_path: Path
future: RunwayFutureDefinitionModel
ignore_git_branch: bool
runway_version: Optional[SpecifierSet]
tests: List[RunwayTestDefinition[Any]]
variables: RunwayVariablesDefinition
_data: RunwayConfigDefinitionModel
[docs] def __init__(
self, data: RunwayConfigDefinitionModel, *, path: Optional[Path] = None
) -> None:
"""Instantiate class.
Args:
data: The data model of the config file.
path: Path to the config file.
"""
super().__init__(data, path=path)
self.deployments = [
RunwayDeploymentDefinition(d) for d in self._data.deployments
]
self.future = self._data.future
self.ignore_git_branch = self._data.ignore_git_branch
self.runway_version = self._data.runway_version
self.tests = [RunwayTestDefinition(t) for t in self._data.tests]
self.variables = RunwayVariablesDefinition(self._data.variables)
[docs] @classmethod
def find_config_file(cls, path: Path) -> Path:
"""Find a config file in the provided path.
Args:
path: The path to search for a config file.
Raises:
ConfigNotFound: Could not find a config file in the provided path.
ValueError: More than one config file found in the provided path.
"""
match = list(path.glob("runway.y*"))
if not match or all(f.name not in cls.ACCEPTED_NAMES for f in match):
match = list(path.parent.glob("runway.y*"))
found = [f for f in match if f.is_file() and f.name in cls.ACCEPTED_NAMES]
if not found:
raise ConfigNotFound(looking_for=cls.ACCEPTED_NAMES, path=path)
if len(found) != 1:
raise ValueError(f"more than one config files found: {found}")
return found[0]
[docs] @classmethod
def parse_file(
cls,
*,
path: Optional[Path] = None,
file_path: Optional[Path] = None,
**kwargs: Any,
) -> RunwayConfig:
"""Parse a YAML file to create a config object.
Args:
path: The path to search for a config file.
file_path: Exact path to a file to parse.
Raises:
ConfigNotFound: Provided config file was not found.
ValueError: path and file_path were both excluded.
"""
if file_path:
if not file_path.is_file():
raise ConfigNotFound(path=file_path)
return cls.parse_obj(
yaml.safe_load(file_path.read_text()), path=file_path, **kwargs
)
if path:
return cls.parse_file(file_path=cls.find_config_file(path), **kwargs)
raise ValueError("must provide path or file_path")
[docs] @classmethod
def parse_obj(cls, obj: Any, *, path: Optional[Path] = None) -> RunwayConfig:
"""Parse a python object into a config object.
Args:
obj: The object to be parsed.
path: Path to the file the object was parsed from.
"""
return cls(RunwayConfigDefinitionModel.parse_obj(obj), path=path)