"""Runway exceptions."""
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
from .utils import DOC_SITE
if TYPE_CHECKING:
from types import ModuleType
from .variables import (
Variable,
VariableValue,
VariableValueConcatenation,
VariableValueLookup,
)
[docs]class RunwayError(Exception):
"""Base class for custom exceptions raised by Runway."""
message: str
"""Error message."""
[docs] def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Instantiate class."""
if self.message:
super().__init__(self.message, *args, **kwargs)
else:
super().__init__(*args, **kwargs)
[docs]class ConfigNotFound(RunwayError):
"""Configuration file could not be found."""
looking_for: List[str]
message: str
path: Path
[docs] def __init__(self, *, looking_for: Optional[List[str]] = None, path: Path) -> None:
"""Instantiate class.
Args:
path: Path where the config file was expected to be found.
looking_for: List of file names that were being looked for.
"""
self.looking_for = looking_for or []
self.path = path
if looking_for:
self.message = (
f"config file not found at path {path}; "
f"looking for one of {looking_for}"
)
else:
self.message = f"config file not found at path {path}"
super().__init__(self.path, self.looking_for)
[docs]class DockerConnectionRefusedError(RunwayError):
"""Docker connection refused.
This can be caused by a number of reasons:
- Docker is not installed.
- Docker service is not running.
- The current user does not have adequate permissions (e.g. not a member of
the ``docker`` group).
"""
[docs] def __init__(self) -> None:
"""Instantiate class."""
self.message = (
"Docker connection refused; install Docker, ensure it is running, "
"and ensure the current user has adequate permissions"
)
super().__init__()
[docs]class DockerExecFailedError(RunwayError):
"""Docker failed when trying to execute a command.
This can be used for ``docker container run``, ``docker container start``
and ``docker exec`` equivalents.
"""
exit_code: int
"""The ``StatusCode`` returned by Docker."""
[docs] def __init__(self, response: Dict[str, Any]) -> None:
"""Instantiate class.
Args:
response: The return value of :meth:`docker.models.containers.Container.wait`,
Docker API's response as a Python dictionary.
This can contain important log information pertinent to troubleshooting
that may not streamed.
"""
self.exit_code = response.get("StatusCode", 1) # we can assume this will be > 0
error = response.get("Error") or {} # value from dict could be NoneType
self.message = error.get("Message", "error message undefined")
super().__init__()
[docs]class FailedLookup(RunwayError):
"""Intermediary Exception to be converted to FailedVariableLookup.
Should be caught by error handling and
:class:`runway.cfngin.exceptions.FailedVariableLookup` raised instead to
construct a proper error message.
"""
cause: Exception
lookup: VariableValueLookup
message: str = "Failed lookup"
[docs] def __init__(
self, lookup: VariableValueLookup, cause: Exception, *args: Any, **kwargs: Any
) -> None:
"""Instantiate class.
Args:
lookup: The variable value lookup that was attempted and
resulted in an exception being raised.
cause: The exception that was raised.
"""
self.cause = cause
self.lookup = lookup
super().__init__(*args, **kwargs)
[docs]class FailedVariableLookup(RunwayError):
"""Lookup could not be resolved.
Raised when an exception is raised when trying to resolve a lookup.
"""
cause: FailedLookup
variable: Variable
message: str
[docs] def __init__(
self, variable: Variable, lookup_error: FailedLookup, *args: Any, **kwargs: Any
) -> None:
"""Instantiate class.
Args:
variable: The variable containing the failed lookup.
lookup_error: The exception that was raised directly before this one.
"""
self.cause = lookup_error
self.variable = variable
self.message = (
f'Could not resolve lookup "{lookup_error.lookup}" '
f'for variable "{variable.name}"'
)
super().__init__(*args, **kwargs)
[docs]class HclParserError(RunwayError):
"""HCL/HCL2 parser error."""
message: str
[docs] def __init__(
self,
exc: Exception,
file_path: Union[Path, str],
parser: Optional[ModuleType] = None,
) -> None:
"""Instantiate class.
Args:
exc: Exception that was raised.
file_path: File that resulted in the error.
parser: The parser what was used to try to parse the file (hcl|hcl2).
"""
self.reason = exc
self.file_path = file_path
if parser:
self.message = (
f"Unable to parse {file_path} as {parser.__name__.upper()}\n\n{exc}"
)
else:
self.message = f"Unable to parse {file_path}\n\n{exc}"
super().__init__()
[docs]class InvalidLookupConcatenation(RunwayError):
"""Invalid return value for a concatenated (chained) lookup.
The return value must be a string when lookups are concatenated.
"""
concatenated_lookups: VariableValueConcatenation[Any]
invalid_lookup: VariableValue
message: str
[docs] def __init__(
self,
invalid_lookup: VariableValue,
concat_lookups: VariableValueConcatenation[Any],
*args: Any,
**kwargs: Any,
) -> None:
"""Instantiate class."""
self.concatenated_lookups = concat_lookups
self.invalid_lookup = invalid_lookup
self.message = (
f"expected return value of type {str} but received "
f'{type(invalid_lookup.value)} for lookup "{invalid_lookup}" '
f'in "{concat_lookups}"'
)
super().__init__(*args, **kwargs)
[docs]class KubectlVersionNotSpecified(RunwayError):
"""kubectl version is required but was not specified.
Version can be specified by using a file or option.
"""
message: str
[docs] def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Instantiate class."""
self.message = (
"kubectl version not specified. Learn how to use Runway to manage kubectl versions "
f"at {DOC_SITE}/page/kubernetes/advanced_features.html#version-management"
)
super().__init__(*args, **kwargs)
[docs]class NpmNotFound(RunwayError):
"""Raised when npm could not be executed or was not found in path."""
message: str
[docs] def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Instantiate class."""
self.message = (
'"npm" not found in path or is not executable; '
"please ensure it is installed correctly"
)
super().__init__(*args, **kwargs)
[docs]class OutputDoesNotExist(RunwayError):
"""Raised when a specific stack output does not exist."""
output: str
"""Name of the CloudFormation Stack's Output that does not exist."""
stack_name: str
"""Name of a CloudFormation Stack."""
[docs] def __init__(self, stack_name: str, output: str, *args: Any, **kwargs: Any) -> None:
"""Instantiate class.
Args:
stack_name: Name of the stack.
output: The output that does not exist.
"""
self.stack_name = stack_name
self.output = output
self.message = f"Output {output} does not exist on stack {stack_name}"
super().__init__(*args, **kwargs)
[docs]class RequiredTagNotFoundError(RunwayError):
"""Required tag not found on resource."""
resource: str
"""An ID or name to identify a resource."""
tag_key: str
"""Key of the tag that could not be found."""
[docs] def __init__(self, resource: str, tag_key: str) -> None:
"""Instantiate class.
Args:
resource: An ID or name to identify a resource.
tag_key: Key of the tag that could not be found.
"""
self.resource = resource
self.tag_key = tag_key
self.message = f"required tag '{tag_key}' not found for {resource}"
super().__init__()
[docs]class UnknownLookupType(RunwayError):
"""Lookup type provided does not match a registered lookup.
Example:
If a lookup of ``${<lookup_type> query}`` is used and ``<lookup_type>``
is not a registered lookup, this exception will be raised.
"""
message: str
[docs] def __init__(self, lookup: VariableValueLookup, *args: Any, **kwargs: Any) -> None:
"""Instantiate class.
Args:
lookup: Variable value lookup that could not find a handler.
"""
self.message = f'Unknown lookup type "{lookup.lookup_name.value}" in "{lookup}"'
super().__init__(*args, **kwargs)
[docs]class UnresolvedVariable(RunwayError):
"""Raised when trying to use a variable before it has been resolved."""
message: str
[docs] def __init__(self, variable: Variable, *args: Any, **kwargs: Any) -> None:
"""Instantiate class.
Args:
variable: The unresolved variable.
"""
self.message = (
f'Attempted to use variable "{variable.name}" before it was resolved'
)
self.variable = variable
super().__init__(*args, **kwargs)
[docs]class UnresolvedVariableValue(RunwayError):
"""Intermediary Exception to be converted to UnresolvedVariable.
Should be caught by error handling and
:class:`runway.cfngin.exceptions.UnresolvedVariable` raised instead to
construct a proper error message.
"""
lookup: VariableValueLookup
message: str = "Unresolved lookup"
[docs] def __init__(self, lookup: VariableValueLookup, *args: Any, **kwargs: Any) -> None:
"""Instantiate class.
Args:
lookup: The variable value lookup that is not resolved.
"""
self.lookup = lookup
super().__init__(*args, **kwargs)
[docs]class VariablesFileNotFound(RunwayError):
"""Defined variables file could not be found."""
file_path: Path
message: str
[docs] def __init__(self, file_path: Path) -> None:
"""Instantiate class.
Args:
file_path: Path where the file was expected to be found.
"""
self.file_path = file_path
self.message = f"defined variables file not found at path {file_path}"
super().__init__(self.file_path)