Source code for runway.context

"""Runway context module."""
import logging
import sys
from distutils.util import strtobool  # pylint: disable=E

from six import string_types

from .cfngin.session_cache import get_session
from .core.components import DeployEnvironment
from .util import cached_property

LOGGER = logging.getLogger(__name__)

[docs]class Context(object): """Runway execution context.""" # TODO implement propper keyword-only args when dropping python 2 # def __init__(self, # *: Any, # command: Optional[str] = None, # deploy_environment: Optional[DeployEnvironment] = None # ) -> None: def __init__(self, _=None, **kwargs): """Instantiate class. Keywork Arguments: command (Optional[str]): Runway command/action being run. deploy_environment (Optional[DeployEnvironment]): Current deploy environment. """ self.command = kwargs.pop("command", None) self.env = kwargs.pop("deploy_environment", DeployEnvironment()) self.debug = self.env.debug # TODO remove after IaC tools support AWS SSO self.__inject_profile_credentials() @property def boto3_credentials(self): """Return a dict of boto3 credentials.""" return {key.lower(): value for key, value in self.current_aws_creds.items()} @property def current_aws_creds(self): """AWS credentials from self.env_vars. Returns: Dict[str, str] """ return self.env.aws_credentials @cached_property def env_name(self): """Get name from deploy environment [DEPRECATED].""" return @property def env_region(self): # type: () -> str """Get or set the current AWS region [DEPRECATED].""" return self.env.aws_region @env_region.setter def env_region(self, region): # type: (str) -> None """Set the AWS region [DEPRECATED].""" self.env.aws_region = region @property def env_root(self): """Get environment root directory [DEPRECATED].""" return str(self.env.root_dir) @property def env_vars(self): """Get environment variables [DEPRECATED].""" return self.env.vars @cached_property def no_color(self): # type: () -> bool """Wether to explicitly disable color output. Primarily applies to IaC being wrapped by Runway. Returns: bool """ colorize = self.env.vars.get("RUNWAY_COLORIZE") # explicitly enable/disable try: if isinstance(colorize, bool): # catch False return not colorize if colorize and isinstance(colorize, string_types): return not strtobool(colorize) except ValueError: pass # likely invalid RUNWAY_COLORIZE value return not sys.stdout.isatty() @property def is_interactive(self): # type: () -> bool """Wether the user should be prompted or not. Determined by the existed of ``CI`` in the environment. Returns: bool """ return not @property def is_noninteractive(self): # type: () -> bool """Wether the user should be prompted or not. Determined by the existed of ``CI`` in the environment. Inverse of ``is_interactive`` property. Returns: bool """ return @property def is_python3(self): # type: () -> bool """Wether running in Python 3 or not. Used for Python compatability decisions. Returns: bool """ return sys.version_info.major > 2 @property def use_concurrent(self): # type: () -> bool """Wether to use concurrent.futures or not. Noninteractive is required for concurrent execution to prevent weird user-input behavior. Python 3 is required because backported futures has issues with ProcessPoolExecutor. Returns: bool """ if self.is_noninteractive: if self.is_python3: return True LOGGER.warning("Parallel execution disabled; Python 3+ is required") LOGGER.warning("Parallel execution disabled; not running in CI mode") return False
[docs] def copy(self): # type: () -> Context """Copy the contents of this object into a new instance. Returns: Context: New instance with the same contents. """ LOGGER.debug("creating a copy of Runway context...") return self.__class__(command=self.command, deploy_environment=self.env.copy())
[docs] def echo_detected_environment(self): # type: () -> None """Print a helper note about how the environment was determined.""" self.env.log_name()
[docs] def get_session(self, profile=None, region=None): """Create a thread-safe boto3 session. Args: profile (Optional[str]): The profile for the session. region (Optional[str]): The region for the session. Returns: :class:`boto3.session.Session`: A thread-safe boto3 session. """ kwargs = {} # save to var so its not calculated multiple times creds = self.boto3_credentials if profile: LOGGER.verbose('creating AWS session using profile "%s"...', profile) kwargs["profile"] = profile elif creds: LOGGER.verbose( "creating AWS session using credentials from the environment..." ) kwargs.update( { "access_key": creds.get("aws_access_key_id"), "secret_key": creds.get("aws_secret_access_key"), "session_token": creds.get("aws_session_token"), } ) return get_session(region=region or self.env.aws_region, **kwargs)
# TODO remove after IaC tools support AWS SSO def __inject_profile_credentials(self): # cov: ignore """Inject AWS credentials into self.env_vars if using an AWS profile. This is to enable support of AWS SSO profiles until all IaC tools that Runway wraps supports these types of profiles. """ if self.current_aws_creds or not self.env.aws_profile: return creds = ( self.get_session(profile=self.env.aws_profile) .get_credentials() .get_frozen_credentials() ) self.env.vars["AWS_ACCESS_KEY_ID"] = creds.access_key self.env.vars["AWS_SECRET_ACCESS_KEY"] = creds.secret_key if creds.token: self.env.vars["AWS_SESSION_TOKEN"] = creds.token