"""Runway config models."""
# pylint: disable=no-self-argument
from __future__ import annotations
import locale
import logging
from pathlib import Path
from typing import (
TYPE_CHECKING,
Any,
Callable,
Dict,
Generator,
List,
Optional,
Type,
TypeVar,
Union,
cast,
)
import yaml
from packaging.specifiers import InvalidSpecifier, SpecifierSet
from pydantic import Extra, Field, Protocol, root_validator, validator
from typing_extensions import Literal
from .. import utils
from ..base import ConfigProperty
from ..utils import RUNWAY_LOOKUP_STRING_ERROR, RUNWAY_LOOKUP_STRING_REGEX
from ._builtin_tests import (
CfnLintRunwayTestArgs,
CfnLintRunwayTestDefinitionModel,
RunwayTestDefinitionModel,
ScriptRunwayTestArgs,
ScriptRunwayTestDefinitionModel,
ValidRunwayTestTypeValues,
YamlLintRunwayTestDefinitionModel,
)
if TYPE_CHECKING:
from pydantic import BaseModel
Model = TypeVar("Model", bound=BaseModel)
LOGGER = logging.getLogger(__name__)
RunwayEnvironmentsType = Dict[str, Union[bool, List[str], str]]
RunwayEnvironmentsUnresolvedType = Union[Dict[str, Union[bool, List[str], str]], str]
RunwayEnvVarsType = Dict[str, Union[List[str], str]]
RunwayEnvVarsUnresolvedType = Union[RunwayEnvVarsType, str]
RunwayModuleTypeTypeDef = Literal[
"cdk", "cloudformation", "kubernetes", "serverless", "static", "terraform"
]
__all__ = [
"CfnLintRunwayTestArgs",
"CfnLintRunwayTestDefinitionModel",
"RUNWAY_LOOKUP_STRING_ERROR",
"RUNWAY_LOOKUP_STRING_REGEX",
"RunwayAssumeRoleDefinitionModel",
"RunwayConfigDefinitionModel",
"RunwayDeploymentDefinitionModel",
"RunwayDeploymentRegionDefinitionModel",
"RunwayEnvironmentsType",
"RunwayEnvironmentsUnresolvedType",
"RunwayEnvVarsType",
"RunwayEnvVarsUnresolvedType",
"RunwayFutureDefinitionModel",
"RunwayModuleDefinitionModel",
"RunwayModuleTypeTypeDef",
"RunwayTestDefinitionModel",
"RunwayVariablesDefinitionModel",
"RunwayVersionField",
"ScriptRunwayTestArgs",
"ScriptRunwayTestDefinitionModel",
"ValidRunwayTestTypeValues",
"YamlLintRunwayTestDefinitionModel",
]
[docs]class RunwayAssumeRoleDefinitionModel(ConfigProperty):
"""Model for a Runway assume role definition."""
arn: Optional[str] = Field(
default=None,
title="IAM Role ARN",
description="The ARN of the AWS IAM role to be assumed. (supports lookups)",
)
duration: Union[int, str] = Field(
default=3600,
description="The duration, in seconds, of the role session. (supports lookups)",
ge=900, # applies to int json schema only
le=43_200, # applies to int json schema only
regex=RUNWAY_LOOKUP_STRING_REGEX, # applies to str json schema only
)
post_deploy_env_revert: bool = Field(
default=False,
title="Post Deployment Environment Revert",
description="Revert the credentials stored in environment variables to "
"what they were prior to execution after the deployment finished processing. "
"(supports lookups)",
)
session_name: str = Field(
default="runway",
description="An identifier for the assumed role session. (supports lookups)",
)
[docs] class Config(ConfigProperty.Config):
"""Model configuration."""
extra = Extra.forbid
schema_extra: Dict[str, Any] = {
"description": "Used to defined a role to assume while Runway is "
"processing each module.",
"examples": [
{"arn": "arn:aws:iam::123456789012:role/name"},
{
"arn": "${var role_arn.${env DEPLOY_ENVIRONMENT}}",
"duration": 9001,
"post_deploy_env_revert": True,
"session_name": "runway-example",
},
],
}
title = "Runway Deployment.assume_role Definition"
@validator("arn")
def _convert_arn_null_value(cls, v: Optional[str]) -> Optional[str]:
"""Convert a "nul" string into type(None)."""
null_strings = ["null", "none", "undefined"]
return None if isinstance(v, str) and v.lower() in null_strings else v
@validator("duration", pre=True)
def _validate_duration(cls, v: Union[int, str]) -> Union[int, str]:
"""Validate duration is within the range allowed by AWS."""
if isinstance(v, str):
return v
if v < 900:
raise ValueError("duration must be greater than or equal to 900")
if v > 43_200:
raise ValueError("duration must be less than or equal to 43,200")
return v
_validate_string_is_lookup = cast(
"classmethod[Callable[..., Any]]",
validator("duration", allow_reuse=True, pre=True)(
utils.validate_string_is_lookup
),
)
[docs]class RunwayDeploymentRegionDefinitionModel(ConfigProperty):
"""Model for a Runway deployment region definition."""
parallel: Union[List[str], str] = Field(
...,
title="Parallel Regions",
description="An array of AWS Regions to process asynchronously. (supports lookups)",
)
[docs] class Config(ConfigProperty.Config):
"""Model configuration."""
extra = Extra.forbid
schema_extra: Dict[str, Any] = {
"description": "Only supports 'parallel' field.",
"examples": [
{"parallel": ["us-east-1", "us-east-2"]},
{"parallel": "${var regions.${env DEPLOY_ENVIRONMENT}}"},
],
}
title = "Runway Deployment.regions Definition"
_validate_string_is_lookup = cast(
"classmethod[Callable[..., Any]]",
validator("parallel", allow_reuse=True, pre=True)(
utils.validate_string_is_lookup
),
)
[docs]class RunwayDeploymentDefinitionModel(ConfigProperty):
"""Model for a Runway deployment definition."""
account_alias: Optional[str] = Field(
default=None,
description="Used to verify the currently assumed role or credentials. "
"(supports lookups)",
examples=["example-alias", "${var alias.${env DEPLOY_ENVIRONMENT}}"],
)
account_id: Optional[str] = Field(
default=None,
description="Used to verify the currently assumed role or credentials. "
"(supports lookups)",
examples=["123456789012", "${var id.${env DEPLOY_ENVIRONMENT}}"],
)
assume_role: Union[str, RunwayAssumeRoleDefinitionModel] = Field(
default={},
description="Assume a role when processing the deployment. (supports lookups)",
examples=["arn:aws:iam::123456789012:role/name"]
+ cast(
List[Any], RunwayAssumeRoleDefinitionModel.Config.schema_extra["examples"]
),
)
env_vars: RunwayEnvVarsUnresolvedType = Field(
default={},
title="Environment Variables",
description="Additional variables to add to the environment when "
"processing the deployment. (supports lookups)",
examples=[
"${var env_vars.${env DEPLOY_ENVIRONMENT}}",
{
"EXAMPLE_VARIABLE": "value",
"KUBECONFIG": [".kube", "${env DEPLOY_ENVIRONMENT}", "config"],
},
],
)
environments: RunwayEnvironmentsUnresolvedType = Field(
default={},
description="Explicitly enable/disable the deployment for a specific "
"deploy environment, AWS Account ID, and AWS Region combination. "
"Can also be set as a static boolean value. (supports lookups)",
examples=[
"${var envs.${env DEPLOY_ENVIRONMENT}}",
{"dev": "123456789012", "prod": "us-east-1"},
{"dev": True, "prod": False},
{"dev": ["us-east-1"], "prod": ["us-west-2", "ca-central-1"]},
{
"dev": ["123456789012/us-east-1", "123456789012/us-west-2"],
"prod": ["234567890123/us-east-1", "234567890123/us-west-2"],
},
],
)
modules: List[RunwayModuleDefinitionModel] = Field(
..., description="An array of modules to process as part of a deployment."
)
module_options: Union[Dict[str, Any], str] = Field(
default={},
description="Options that are passed directly to the modules within this deployment. "
"(supports lookups)",
examples=[
"${var sampleapp.options.${env DEPLOY_ENVIRONMENT}}",
{"some_option": "value"},
],
)
name: str = Field(
default="unnamed_deployment",
description="The name of the deployment to be displayed in logs and the "
"interactive selection menu.",
)
parallel_regions: Union[List[str], str] = Field(
default=[],
description="An array of AWS Regions to process asynchronously. (supports lookups)",
examples=[
["us-east-1", "us-west-2"],
"${var regions.${dev DEPLOY_ENVIRONMENT}}",
],
)
parameters: Union[Dict[str, Any], str] = Field(
default={},
description="Used to pass variable values to modules in place of an "
"environment configuration file. (supports lookups)",
examples=[
{"namespace": "example-${env DEPLOY_ENVIRONMENT}"},
"${var sampleapp.parameters.${env DEPLOY_ENVIRONMENT}}",
],
)
regions: Union[List[str], str] = Field(
default=[],
description="An array of AWS Regions to process this deployment in. (supports lookups)",
examples=[
["us-east-1", "us-west-2"],
"${var regions.${dev DEPLOY_ENVIRONMENT}}",
]
+ RunwayDeploymentRegionDefinitionModel.Config.schema_extra["examples"],
)
[docs] class Config(ConfigProperty.Config):
"""Model configuration."""
extra = Extra.forbid
title = "Runway Deployment Definition"
@root_validator(pre=True)
def _convert_simple_module(cls, values: Dict[str, Any]) -> Dict[str, Any]:
"""Convert simple modules to dicts."""
modules = values.get("modules", [])
result: List[Dict[str, Any]] = []
for module in modules:
if isinstance(module, str):
result.append({"path": module})
else:
result.append(module)
values["modules"] = result
return values
@root_validator(pre=True)
def _validate_regions(cls, values: Dict[str, Any]) -> Dict[str, Any]:
"""Validate & simplify regions."""
raw_regions = values.get("regions", [])
parallel_regions = values.get("parallel_regions", [])
if all(isinstance(i, str) for i in [raw_regions, parallel_regions]):
raise ValueError(
"unable to validate parallel_regions/regions - both are defined as strings"
)
if any(isinstance(i, str) for i in [raw_regions, parallel_regions]):
return values # one is a lookup so skip the remainder of the checks
if isinstance(raw_regions, list):
regions = raw_regions
else:
regions = RunwayDeploymentRegionDefinitionModel.parse_obj(raw_regions)
if regions and parallel_regions:
raise ValueError("only one of parallel_regions or regions can be defined")
if not regions and not parallel_regions:
raise ValueError("either parallel_regions or regions must be defined")
if isinstance(regions, RunwayDeploymentRegionDefinitionModel):
values["regions"] = []
values["parallel_regions"] = regions.parallel
return values
_validate_string_is_lookup = cast(
"classmethod[Callable[..., Any]]",
validator(
"env_vars",
"environments",
"module_options",
"parallel_regions",
"parameters",
"regions",
allow_reuse=True,
pre=True,
)(utils.validate_string_is_lookup),
)
[docs]class RunwayFutureDefinitionModel(ConfigProperty):
"""Model for the Runway future definition."""
[docs] class Config(ConfigProperty.Config):
"""Model configuration."""
extra = Extra.forbid
schema_extra = {
"description": "Enable features/behaviors that will be become standard "
"ahead of their official release."
}
title = "Runway Future Definition"
[docs]class RunwayModuleDefinitionModel(ConfigProperty):
"""Model for a Runway module definition."""
class_path: Optional[str] = Field(
default=None,
description="Import path to a custom Runway module class. (supports lookups)",
)
env_vars: RunwayEnvVarsUnresolvedType = Field(
default={},
title="Environment Variables",
description="Additional variables to add to the environment when "
"processing the deployment. (supports lookups)",
examples=[
"${var env_vars.${env DEPLOY_ENVIRONMENT}}",
{
"EXAMPLE_VARIABLE": "value",
"KUBECONFIG": [".kube", "${env DEPLOY_ENVIRONMENT}", "config"],
},
],
)
environments: RunwayEnvironmentsUnresolvedType = Field(
default={},
description="Explicitly enable/disable the deployment for a specific "
"deploy environment, AWS Account ID, and AWS Region combination. "
"Can also be set as a static boolean value. (supports lookups)",
examples=[
"${var envs.${env DEPLOY_ENVIRONMENT}}",
{"dev": "123456789012", "prod": "us-east-1"},
{"dev": True, "prod": False},
{"dev": ["us-east-1"], "prod": ["us-west-2", "ca-central-1"]},
{
"dev": ["123456789012/us-east-1", "123456789012/us-west-2"],
"prod": ["234567890123/us-east-1", "234567890123/us-west-2"],
},
],
)
name: str = Field(
default="undefined",
description="The name of the module to be displayed in logs and the "
"interactive selection menu.",
)
options: Union[Dict[str, Any], str] = Field(
default={}, description="Module type specific options. (supports lookups)"
)
parameters: Union[Dict[str, Any], str] = Field(
default={},
description="Used to pass variable values to modules in place of an "
"environment configuration file. (supports lookups)",
examples=[
{"namespace": "example-${env DEPLOY_ENVIRONMENT}"},
"${var sampleapp.parameters.${env DEPLOY_ENVIRONMENT}}",
],
)
path: Optional[Union[str, Path]] = Field(
default=None,
description="Directory (relative to the Runway config file) containing IaC. "
"(supports lookups)",
examples=["./", "sampleapp-${env DEPLOY_ENVIRONMENT}.cfn", "sampleapp.sls"],
)
tags: List[str] = Field(
default=[],
description="Array of values to categorize the module which can be used "
"with the CLI to quickly select a group of modules. "
"This field is only used by the `--tag` CLI option.",
examples=[["type:network", "app:sampleapp"]],
)
type: Optional[RunwayModuleTypeTypeDef] = None
# needs to be last
parallel: List[RunwayModuleDefinitionModel] = Field(
default=[],
description="Array of module definitions that can be executed asynchronously. "
"Incompatible with class_path, path, and type.",
examples=[[{"path": "sampleapp-01.cfn"}, {"path": "sampleapp-02.cfn"}]],
)
[docs] class Config(ConfigProperty.Config):
"""Model configuration."""
extra = Extra.forbid
schema_extra = {
"description": "Defines a directory containing IaC, "
"the parameters to pass in during execution, "
"and any applicable options for the module type.",
}
title = "Runway Module Definition"
use_enum_values = True
@root_validator(pre=True)
def _validate_name(cls, values: Dict[str, Any]) -> Dict[str, Any]:
"""Validate module name."""
if "name" in values:
return values
if "parallel" in values:
values["name"] = "parallel_parent"
return values
if "path" in values:
values["name"] = Path(values["path"]).resolve().name
return values
return values
@root_validator(pre=True)
def _validate_path(cls, values: Dict[str, Any]) -> Dict[str, Any]:
"""Validate path and sets a default value if needed."""
if not values.get("path") and not values.get("parallel"):
values["path"] = Path.cwd()
return values
@validator("parallel", pre=True)
def _validate_parallel(
cls, v: List[Union[Dict[str, Any], str]], values: Dict[str, Any]
) -> List[Dict[str, Any]]:
"""Validate parallel."""
if v and values.get("path"):
raise ValueError("only one of parallel or path can be defined")
result: List[Dict[str, Any]] = []
for mod in v:
if isinstance(mod, str):
result.append({"path": mod})
else:
result.append(mod)
return result
# TODO add regex to schema
_validate_string_is_lookup = cast(
"classmethod[Callable[..., Any]]",
validator(
"env_vars",
"environments",
"options",
"parameters",
allow_reuse=True,
pre=True,
)(utils.validate_string_is_lookup),
)
# https://pydantic-docs.helpmanual.io/usage/postponed_annotations/#self-referencing-models
RunwayModuleDefinitionModel.update_forward_refs()
[docs]class RunwayVariablesDefinitionModel(ConfigProperty):
"""Model for a Runway variable definition."""
file_path: Optional[Path] = Field(
default=None,
title="Variables File Path",
description="Explicit path to a variables file that will be loaded and "
"merged with the variables defined here.",
)
sys_path: Path = Field(
default="./",
description="Directory to use as the root of a relative 'file_path'. "
"If not provided, the current working directory is used.",
)
[docs] class Config(ConfigProperty.Config):
"""Model configuration."""
extra = Extra.allow
schema_extra = {
"description": "A variable definitions for the Runway config file. "
"This is used to resolve the 'var' lookup.",
}
title = "Runway Variables Definition"
_convert_null_values = cast(
"classmethod[Callable[..., Any]]",
validator("*", allow_reuse=True)(utils.convert_null_values),
)
[docs]class RunwayVersionField(SpecifierSet):
"""Extends packaging.specifiers.SpecifierSet for use with pydantic."""
[docs] @classmethod
def __get_validators__(cls) -> Generator[Callable[..., Any], None, None]:
"""Yield one of more validators with will be called to validate the input.
Each validator will receive, as input, the value returned from the previous validator.
"""
yield cls._convert_value
[docs] @classmethod
def __modify_schema__(cls, field_schema: Dict[str, Any]) -> None:
"""Mutate the field schema in place.
This is only called when output JSON schema from a model.
"""
field_schema.update(type="string") # cov: ignore
@classmethod
def _convert_value(cls, v: Union[str, SpecifierSet]) -> RunwayVersionField:
"""Convert runway_version string into SpecifierSet with some value handling.
Args:
v: The value to be converted/validated.
Raises:
ValueError: The provided value is not a valid version specifier set.
"""
if isinstance(v, (cls, SpecifierSet)):
return RunwayVersionField(str(v), prereleases=True)
try:
return RunwayVersionField(v, prereleases=True)
except InvalidSpecifier:
if any(v.startswith(i) for i in ["!", "~", "<", ">", "="]):
raise ValueError(f"{v} is not a valid version specifier set") from None
LOGGER.debug(
"runway_version is not a valid version specifier; trying as an exact version",
exc_info=True,
)
return RunwayVersionField("==" + v, prereleases=True)
[docs]class RunwayConfigDefinitionModel(ConfigProperty):
"""Runway configuration definition model."""
deployments: List[RunwayDeploymentDefinitionModel] = Field(
default=[], description="Array of Runway deployments definitions."
)
future: RunwayFutureDefinitionModel = RunwayFutureDefinitionModel()
ignore_git_branch: bool = Field(
default=False,
description="Optionally exclude the git branch name when determining the "
"current deploy environment.",
)
runway_version: Optional[RunwayVersionField] = Field(
default=">1.10",
description="Define the versions of Runway that can be used with this "
"configuration file.",
examples=['"<2.0.0"', '"==1.14.0"', '">=1.14.0,<2.0.0"'],
)
tests: List[RunwayTestDefinitionModel] = Field(
default=[],
description="Array of Runway test definitions that are executed with the 'test' command.",
)
variables: RunwayVariablesDefinitionModel = RunwayVariablesDefinitionModel()
[docs] class Config(ConfigProperty.Config):
"""Model configuration."""
extra = Extra.forbid
schema_extra = {
"description": "Configuration file for use with Runway.",
}
title = "Runway Configuration File"
validate_all = True
validate_assignment = True
@root_validator(pre=True)
def _add_deployment_names(cls, values: Dict[str, Any]) -> Dict[str, Any]:
"""Add names to deployments that are missing them."""
deployments = values.get("deployments", [])
for i, deployment in enumerate(deployments):
if not deployment.get("name"):
deployment["name"] = f"deployment_{i + 1}"
values["deployments"] = deployments
return values
[docs] @classmethod
def parse_file(
cls: Type[Model],
path: Union[str, Path],
*,
content_type: Optional[str] = None,
encoding: str = "utf8",
proto: Optional[Protocol] = None,
allow_pickle: bool = False,
) -> Model:
"""Parse a file."""
return cast(
"Model",
cls.parse_raw(
Path(path).read_text(
encoding=locale.getpreferredencoding(do_setlocale=False)
),
content_type=content_type, # type: ignore
encoding=encoding,
proto=proto, # type: ignore
allow_pickle=allow_pickle,
),
)
[docs] @classmethod
def parse_raw(
cls: Type[Model],
b: Union[bytes, str],
*,
content_type: Optional[str] = None, # pylint: disable=unused-argument
encoding: str = "utf8", # pylint: disable=unused-argument
proto: Optional[Protocol] = None, # pylint: disable=unused-argument
allow_pickle: bool = False, # pylint: disable=unused-argument
) -> Model:
"""Parse raw data."""
return cast("Model", cls.parse_obj(yaml.safe_load(b)))
# https://pydantic-docs.helpmanual.io/usage/postponed_annotations/#self-referencing-models
RunwayDeploymentDefinitionModel.update_forward_refs()