samtranslator/model/preferences/deployment_preference.py (46 lines of code) (raw):
from collections import namedtuple
from samtranslator.model.exceptions import InvalidResourceException
"""
:param deployment_type: There are two types of deployments at the moment: Linear and Canary.
There is a default percentage of traffic that is routed to the new function's version for a 30 minute bake
period after which the alias will 100% be routed to the new function version.
Linear deployment type means that every 10 minutes 10% more traffic will be routed to the new function
version.
:param pre_traffic_hook: A lambda function reference that will be used to test the new function version before
any traffic is shifted to it at all. If his pre_traffic_hook fails the lambda deployment stops leaving
nothing unchanged for the customer's production traffic (lambda alias still pointed ot the old version)
:param post_traffic_hook: A lambda function reference that will be used to test the new function version after
all traffic has been shifted for the alias to be 100% pointing to the new function version. If this test
fails CodeDeploy is in charge of rolling back the deployment meaning 100% shifting traffic back to the old
version.
:param alarms: A list of Cloudwatch Alarm references that if ever in the alarm state during a deployment (or
before a deployment starts) cause the deployment to fail and rollback.
:param role: An IAM role ARN that CodeDeploy will use for traffic shifting, an IAM role will not be created if
this is supplied
:param enabled: Whether this deployment preference is enabled (true by default)
:param trigger_configurations: Information about triggers associated with the deployment group. Duplicates are
not allowed.
"""
DeploymentPreferenceTuple = namedtuple(
"DeploymentPreferenceTuple",
[
"deployment_type",
"pre_traffic_hook",
"post_traffic_hook",
"alarms",
"enabled",
"role",
"trigger_configurations",
"condition",
],
)
class DeploymentPreference(DeploymentPreferenceTuple):
"""
The DeploymentPreference object representing the customer's lambda deployment preferences.
Each parameter controls what happens whenever a customer wants to update their lambda function.
The data is "immutable".
"""
@classmethod
def from_dict(cls, logical_id, deployment_preference_dict, condition=None): # type: ignore[no-untyped-def]
"""
:param logical_id: the logical_id of the resource that owns this deployment preference
:param deployment_preference_dict: the dict object taken from the SAM template
:param condition: condition on this deployment preference
:return:
"""
enabled = deployment_preference_dict.get("Enabled", True)
enabled = False if enabled in ["false", "False"] else enabled
if not enabled:
return DeploymentPreference(None, None, None, None, False, None, None, None)
if "Type" not in deployment_preference_dict:
raise InvalidResourceException(logical_id, "'DeploymentPreference' is missing required Property 'Type'")
deployment_type = deployment_preference_dict["Type"]
hooks = deployment_preference_dict.get("Hooks", {})
if not isinstance(hooks, dict):
raise InvalidResourceException(
logical_id, "'Hooks' property of 'DeploymentPreference' must be a dictionary"
)
pre_traffic_hook = hooks.get("PreTraffic", None)
post_traffic_hook = hooks.get("PostTraffic", None)
alarms = deployment_preference_dict.get("Alarms", None)
role = deployment_preference_dict.get("Role", None)
trigger_configurations = deployment_preference_dict.get("TriggerConfigurations", None)
passthrough_condition = deployment_preference_dict.get("PassthroughCondition", False)
return DeploymentPreference(
deployment_type,
pre_traffic_hook,
post_traffic_hook,
alarms,
enabled,
role,
trigger_configurations,
condition if passthrough_condition else None,
)