"""Base classes for runway modules."""
from __future__ import annotations
import logging
import subprocess
from pathlib import Path
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union, cast
from ..exceptions import NpmNotFound
from ..utils import which
from .utils import NPM_BIN, format_npm_command_for_logging, use_npm_ci
if TYPE_CHECKING:
from .._logging import PrefixAdaptor, RunwayLogger
from ..context import RunwayContext
LOGGER = cast("RunwayLogger", logging.getLogger(__name__))
[docs]class RunwayModule:
"""Base class for Runway modules."""
ctx: RunwayContext
explicitly_enabled: Optional[bool]
logger: Union[PrefixAdaptor, RunwayLogger]
name: str
options: Union[Dict[str, Any], ModuleOptions]
region: str
[docs] def __init__(
self,
context: RunwayContext,
*,
explicitly_enabled: Optional[bool] = False,
logger: RunwayLogger = LOGGER,
module_root: Path,
name: Optional[str] = None,
options: Optional[Union[Dict[str, Any], ModuleOptions]] = None,
parameters: Optional[Dict[str, Any]] = None,
**_: Any,
) -> None:
"""Instantiate class.
Args:
context: Runway context object for the current session.
explicitly_enabled: Whether or not the module is explicitly enabled.
This is can be set in the event that the current environment being
deployed to matches the defined environments of the module/deployment.
logger: Used to write logs.
module_root: Root path of the module.
name: Name of the module.
options: Options passed to the module class from the config as ``options``
or ``module_options`` if coming from the deployment level.
parameters: Values to pass to the underlying infrastructure as code
tool that will alter the resulting infrastructure being deployed.
Used to templatize IaC.
"""
self.ctx = context
self.explicitly_enabled = explicitly_enabled
self.logger = logger
self.name = name or module_root.name
self.options = options or {}
self.parameters = parameters or {}
self.path = module_root
self.region = context.env.aws_region
[docs] def deploy(self) -> None:
"""Abstract method called when running deploy."""
raise NotImplementedError("You must implement the deploy() method yourself!")
[docs] def destroy(self) -> None:
"""Abstract method called when running destroy."""
raise NotImplementedError("You must implement the destroy() method yourself!")
[docs] def init(self) -> None:
"""Abstract method called when running init."""
raise NotImplementedError("You must implement the init() method yourself!")
[docs] def plan(self) -> None:
"""Abstract method called when running plan."""
raise NotImplementedError("You must implement the plan() method yourself!")
[docs] def __getitem__(self, key: str) -> Any:
"""Make the object subscriptable.
Args:
key: Attribute to get.
"""
return getattr(self, key)
[docs]class RunwayModuleNpm(RunwayModule): # pylint: disable=abstract-method
"""Base class for Runway modules that use npm."""
[docs] def __init__(
self,
context: RunwayContext,
*,
explicitly_enabled: Optional[bool] = False,
logger: RunwayLogger = LOGGER,
module_root: Path,
name: Optional[str] = None,
options: Optional[Union[Dict[str, Any], ModuleOptions]] = None,
parameters: Optional[Dict[str, Any]] = None,
**_: Any,
) -> None:
"""Instantiate class.
Args:
context: Runway context object for the current session.
explicitly_enabled: Whether or not the module is explicitly enabled.
This is can be set in the event that the current environment being
deployed to matches the defined environments of the module/deployment.
logger: Used to write logs.
module_root: Root path of the module.
name: Name of the module.
options: Options passed to the module class from the config as ``options``
or ``module_options`` if coming from the deployment level.
parameters: Values to pass to the underlying infrastructure as code
tool that will alter the resulting infrastructure being deployed.
Used to templatize IaC.
"""
super().__init__(
context,
explicitly_enabled=explicitly_enabled,
logger=logger,
module_root=module_root,
name=name,
options=options,
parameters=parameters,
)
self.check_for_npm(logger=self.logger) # fail fast
self.warn_on_boto_env_vars(self.ctx.env.vars, logger=logger)
[docs] def log_npm_command(self, command: List[str]) -> None:
"""Log an npm command that is going to be run.
Args:
command: List that will be passed into a subprocess.
"""
self.logger.debug("node command: %s", format_npm_command_for_logging(command))
[docs] def npm_install(self) -> None:
"""Run ``npm install``."""
if self.options.get("skip_npm_ci"):
self.logger.info("skipped npm ci/npm install")
return
cmd = [NPM_BIN, "<place-holder>"]
if self.ctx.no_color:
cmd.append("--no-color")
if self.ctx.is_noninteractive and use_npm_ci(self.path):
self.logger.info("running npm ci...")
cmd[1] = "ci"
else:
self.logger.info("running npm install...")
cmd[1] = "install"
subprocess.check_call(cmd)
[docs] def package_json_missing(self) -> bool:
"""Check for the existence for a package.json file in the module.
Returns:
bool: True if the file was not found.
"""
if not (self.path / "package.json").is_file():
self.logger.debug("module is missing package.json")
return True
return False
[docs] @staticmethod
def check_for_npm(
*, logger: Union[logging.Logger, PrefixAdaptor, RunwayLogger] = LOGGER
) -> None:
"""Ensure npm is installed and in the current path.
Args:
logger: Optionally provide a custom logger to use.
"""
if not which("npm"):
logger.error(
'"npm" not found in path or is not executable; '
"please ensure it is installed correctly"
)
raise NpmNotFound
[docs] @staticmethod
def warn_on_boto_env_vars(
env_vars: Dict[str, str],
*,
logger: Union[logging.Logger, PrefixAdaptor, RunwayLogger] = LOGGER,
) -> None:
"""Inform user if boto-specific environment variables are in use.
Args:
env_vars: Environment variables to check.
logger: Optionally provide a custom logger to use.
"""
# https://github.com/serverless/serverless/issues/2151#issuecomment-255646512
if env_vars.get("AWS_DEFAULT_PROFILE") and not env_vars.get("AWS_PROFILE"):
logger.warning(
"AWS_DEFAULT_PROFILE environment variable is set "
"during use of nodejs-based module and AWS_PROFILE is "
"not set -- you likely want to set AWS_PROFILE instead"
)
[docs]class ModuleOptions:
"""Base class for Runway module options."""
[docs] def get(self, name: str, default: Any = None) -> Any:
"""Get a value or return the default."""
return getattr(self, name, default)
[docs] def __eq__(self, other: Any) -> bool:
"""Assess equality."""
if isinstance(other, self.__class__):
return self.__dict__ == other.__dict__
return False