sam_resources.py

""" SAM macro definitions """
from six import string_types

import samtranslator.model.eventsources
import samtranslator.model.eventsources.pull
import samtranslator.model.eventsources.push
import samtranslator.model.eventsources.cloudwatchlogs
from .api.api_generator import ApiGenerator
from .api.http_api_generator import HttpApiGenerator
from .s3_utils.uri_parser import construct_s3_location_object
from .tags.resource_tagging import get_tag_list
from samtranslator.model import (PropertyType, SamResourceMacro,
                                 ResourceTypeResolver)
from samtranslator.model.apigateway import ApiGatewayDeployment, ApiGatewayStage
from samtranslator.model.apigatewayv2 import ApiGatewayV2Stage
from samtranslator.model.cloudformation import NestedStack
from samtranslator.model.dynamodb import DynamoDBTable
from samtranslator.model.exceptions import (InvalidEventException,
                                            InvalidResourceException)
from samtranslator.model.function_policies import FunctionPolicies, PolicyTypes
from samtranslator.model.iam import IAMRole, IAMRolePolicies
from samtranslator.model.lambda_ import (LambdaFunction, LambdaVersion, LambdaAlias,
                                         LambdaLayerVersion)
from samtranslator.model.types import dict_of, is_str, is_type, list_of, one_of, any_type
from samtranslator.translator import logical_id_generator
from samtranslator.translator.arn_generator import ArnGenerator
from samtranslator.model.intrinsics import is_intrinsic_if, is_intrinsic_no_value


class SamFunction(SamResourceMacro):
    """SAM function macro.
    """

    resource_type = 'AWS::Serverless::Function'
    property_types = {
        'FunctionName': PropertyType(False, one_of(is_str(), is_type(dict))),
        'Handler': PropertyType(True, is_str()),
        'Runtime': PropertyType(True, is_str()),
        'CodeUri': PropertyType(False, one_of(is_str(), is_type(dict))),
        'InlineCode': PropertyType(False, one_of(is_str(), is_type(dict))),
        'DeadLetterQueue': PropertyType(False, is_type(dict)),
        'Description': PropertyType(False, is_str()),
        'MemorySize': PropertyType(False, is_type(int)),
        'Timeout': PropertyType(False, is_type(int)),
        'VpcConfig': PropertyType(False, is_type(dict)),
        'Role': PropertyType(False, is_str()),
        'Policies': PropertyType(False, one_of(is_str(), list_of(one_of(is_str(), is_type(dict), is_type(dict))))),
        'PermissionsBoundary': PropertyType(False, is_str()),
        'Environment': PropertyType(False, dict_of(is_str(), is_type(dict))),
        'Events': PropertyType(False, dict_of(is_str(), is_type(dict))),
        'Tags': PropertyType(False, is_type(dict)),
        'Tracing': PropertyType(False, one_of(is_type(dict), is_str())),
        'KmsKeyArn': PropertyType(False, one_of(is_type(dict), is_str())),
        'DeploymentPreference': PropertyType(False, is_type(dict)),
        'ReservedConcurrentExecutions': PropertyType(False, any_type()),
        'Layers': PropertyType(False, list_of(one_of(is_str(), is_type(dict)))),

        # Intrinsic functions in value of Alias property are not supported, yet
        'AutoPublishAlias': PropertyType(False, one_of(is_str())),
        'VersionDescription': PropertyType(False, is_str()),
        'ProvisionedConcurrencyConfig': PropertyType(False, is_type(dict)),
    }
    event_resolver = ResourceTypeResolver(samtranslator.model.eventsources, samtranslator.model.eventsources.pull,
                                          samtranslator.model.eventsources.push,
                                          samtranslator.model.eventsources.cloudwatchlogs)

    # DeadLetterQueue
    dead_letter_queue_policy_actions = {'SQS': 'sqs:SendMessage', 'SNS': 'sns:Publish'}

    # Customers can refer to the following properties of SAM function
    referable_properties = {
        "Alias": LambdaAlias.resource_type,
        "Version": LambdaVersion.resource_type,
    }

    def resources_to_link(self, resources):
        try:
            return {
                'event_resources': self._event_resources_to_link(resources)
            }
        except InvalidEventException as e:
            raise InvalidResourceException(self.logical_id, e.message)

    def to_cloudformation(self, **kwargs):
        """Returns the Lambda function, role, and event resources to which this SAM Function corresponds.

        :param dict kwargs: already-converted resources that may need to be modified when converting this \
        macro to pure CloudFormation
        :returns: a list of vanilla CloudFormation Resources, to which this Function expands
        :rtype: list
        """
        resources = []
        intrinsics_resolver = kwargs["intrinsics_resolver"]
        mappings_resolver = kwargs.get("mappings_resolver", None)

        if self.DeadLetterQueue:
            self._validate_dlq()

        lambda_function = self._construct_lambda_function()
        resources.append(lambda_function)

        if self.ProvisionedConcurrencyConfig:
            if not self.AutoPublishAlias:
                raise InvalidResourceException(self.logical_id, "To set ProvisionedConcurrencyConfig "
                                               "AutoPublishALias must be defined on the function")

        lambda_alias = None
        if self.AutoPublishAlias:
            alias_name = self._get_resolved_alias_name("AutoPublishAlias", self.AutoPublishAlias, intrinsics_resolver)
            lambda_version = self._construct_version(lambda_function, intrinsics_resolver=intrinsics_resolver)
            lambda_alias = self._construct_alias(alias_name, lambda_function, lambda_version)
            resources.append(lambda_version)
            resources.append(lambda_alias)

        if self.DeploymentPreference:
            self._validate_deployment_preference_and_add_update_policy(kwargs.get('deployment_preference_collection',
                                                                                  None),
                                                                       lambda_alias, intrinsics_resolver,
                                                                       mappings_resolver)

        managed_policy_map = kwargs.get('managed_policy_map', {})
        if not managed_policy_map:
            raise Exception('Managed policy map is empty, but should not be.')

        execution_role = None
        if lambda_function.Role is None:
            execution_role = self._construct_role(managed_policy_map)
            lambda_function.Role = execution_role.get_runtime_attr('arn')
            resources.append(execution_role)

        try:
            resources += self._generate_event_resources(lambda_function, execution_role, kwargs['event_resources'],
                                                        lambda_alias=lambda_alias)
        except InvalidEventException as e:
            raise InvalidResourceException(self.logical_id, e.message)

        return resources

    def _get_resolved_alias_name(self, property_name, original_alias_value, intrinsics_resolver):
        """
        Alias names can be supplied as an intrinsic function. This method tries to extract alias name from a reference
        to a parameter. If it cannot completely resolve (ie. if a complex intrinsic function was used), then this
        method raises an exception. If alias name is just a plain string, it will return as is

        :param dict or string original_alias_value: Value of Alias property as provided by the customer
        :param samtranslator.intrinsics.resolver.IntrinsicsResolver intrinsics_resolver: Instance of the resolver that
            knows how to resolve parameter references
        :return string: Alias name
        :raises InvalidResourceException: If the value is a complex intrinsic function that cannot be resolved
        """

        # Try to resolve.
        resolved_alias_name = intrinsics_resolver.resolve_parameter_refs(original_alias_value)

        if not isinstance(resolved_alias_name, string_types):
            # This is still a dictionary which means we are not able to completely resolve intrinsics
            raise InvalidResourceException(self.logical_id,
                                           "'{}' must be a string or a Ref to a template parameter"
                                           .format(property_name))

        return resolved_alias_name

    def _construct_lambda_function(self):
        """Constructs and returns the Lambda function.

        :returns: a list containing the Lambda function and execution role resources
        :rtype: list
        """
        lambda_function = LambdaFunction(self.logical_id, depends_on=self.depends_on,
                                         attributes=self.resource_attributes)

        if self.FunctionName:
            lambda_function.FunctionName = self.FunctionName

        lambda_function.Handler = self.Handler
        lambda_function.Runtime = self.Runtime
        lambda_function.Description = self.Description
        lambda_function.MemorySize = self.MemorySize
        lambda_function.Timeout = self.Timeout
        lambda_function.VpcConfig = self.VpcConfig
        lambda_function.Role = self.Role
        lambda_function.Environment = self.Environment
        lambda_function.Code = self._construct_code_dict()
        lambda_function.KmsKeyArn = self.KmsKeyArn
        lambda_function.ReservedConcurrentExecutions = self.ReservedConcurrentExecutions
        lambda_function.Tags = self._construct_tag_list(self.Tags)
        lambda_function.Layers = self.Layers

        if self.Tracing:
            lambda_function.TracingConfig = {"Mode": self.Tracing}

        if self.DeadLetterQueue:
            lambda_function.DeadLetterConfig = {"TargetArn": self.DeadLetterQueue['TargetArn']}

        return lambda_function

    def _construct_role(self, managed_policy_map):
        """Constructs a Lambda execution role based on this SAM function's Policies property.

        :returns: the generated IAM Role
        :rtype: model.iam.IAMRole
        """
        execution_role = IAMRole(self.logical_id + 'Role', attributes=self.get_passthrough_resource_attributes())
        execution_role.AssumeRolePolicyDocument = IAMRolePolicies.lambda_assume_role_policy()

        managed_policy_arns = [ArnGenerator.generate_aws_managed_policy_arn('service-role/AWSLambdaBasicExecutionRole')]
        if self.Tracing:
            managed_policy_arns.append(ArnGenerator.generate_aws_managed_policy_arn('AWSXrayWriteOnlyAccess'))

        function_policies = FunctionPolicies({"Policies": self.Policies},
                                             # No support for policy templates in the "core"
                                             policy_template_processor=None)
        policy_documents = []

        if self.DeadLetterQueue:
            policy_documents.append(IAMRolePolicies.dead_letter_queue_policy(
                self.dead_letter_queue_policy_actions[self.DeadLetterQueue['Type']],
                self.DeadLetterQueue['TargetArn']))

        for index, policy_entry in enumerate(function_policies.get()):

            if policy_entry.type is PolicyTypes.POLICY_STATEMENT:

                if is_intrinsic_if(policy_entry.data):

                    intrinsic_if = policy_entry.data
                    then_statement = intrinsic_if["Fn::If"][1]
                    else_statement = intrinsic_if["Fn::If"][2]

                    if not is_intrinsic_no_value(then_statement):
                        then_statement = {
                            'PolicyName': execution_role.logical_id + 'Policy' + str(index),
                            'PolicyDocument': then_statement
                        }
                        intrinsic_if["Fn::If"][1] = then_statement

                    if not is_intrinsic_no_value(else_statement):
                        else_statement = {
                            'PolicyName': execution_role.logical_id + 'Policy' + str(index),
                            'PolicyDocument': else_statement
                        }
                        intrinsic_if["Fn::If"][2] = else_statement

                    policy_documents.append(intrinsic_if)

                else:
                    policy_documents.append({
                        'PolicyName': execution_role.logical_id + 'Policy' + str(index),
                        'PolicyDocument': policy_entry.data
                    })

            elif policy_entry.type is PolicyTypes.MANAGED_POLICY:

                # There are three options:
                #   Managed Policy Name (string): Try to convert to Managed Policy ARN
                #   Managed Policy Arn (string): Insert it directly into the list
                #   Intrinsic Function (dict): Insert it directly into the list
                #
                # When you insert into managed_policy_arns list, de-dupe to prevent same ARN from showing up twice
                #

                policy_arn = policy_entry.data
                if isinstance(policy_entry.data, string_types) and policy_entry.data in managed_policy_map:
                    policy_arn = managed_policy_map[policy_entry.data]

                # De-Duplicate managed policy arns before inserting. Mainly useful
                # when customer specifies a managed policy which is already inserted
                # by SAM, such as AWSLambdaBasicExecutionRole
                if policy_arn not in managed_policy_arns:
                    managed_policy_arns.append(policy_arn)
            else:
                # Policy Templates are not supported here in the "core"
                raise InvalidResourceException(
                    self.logical_id,
                    "Policy at index {} in the 'Policies' property is not valid".format(index))

        execution_role.ManagedPolicyArns = list(managed_policy_arns)
        execution_role.Policies = policy_documents or None
        execution_role.PermissionsBoundary = self.PermissionsBoundary

        return execution_role

    def _validate_dlq(self):
        """Validates whether the DeadLetterQueue LogicalId is validation
        :raise: InvalidResourceException
        """
        # Validate required logical ids
        valid_dlq_types = str(list(self.dead_letter_queue_policy_actions.keys()))
        if not self.DeadLetterQueue.get('Type') or not self.DeadLetterQueue.get('TargetArn'):
            raise InvalidResourceException(self.logical_id,
                                           "'DeadLetterQueue' requires Type and TargetArn properties to be specified"
                                           .format(valid_dlq_types))

        # Validate required Types
        if not self.DeadLetterQueue['Type'] in self.dead_letter_queue_policy_actions:
            raise InvalidResourceException(self.logical_id,
                                           "'DeadLetterQueue' requires Type of {}".format(valid_dlq_types))

    def _event_resources_to_link(self, resources):
        event_resources = {}
        if self.Events:
            for logical_id, event_dict in self.Events.items():
                try:
                    event_source = self.event_resolver.resolve_resource_type(event_dict).from_dict(
                        self.logical_id + logical_id, event_dict, logical_id)
                except (TypeError, AttributeError) as e:
                    raise InvalidEventException(logical_id, "{}".format(e))
                event_resources[logical_id] = event_source.resources_to_link(resources)
        return event_resources

    @staticmethod
    def order_events(event):
        """
        Helper method for sorting Function Events. Returns a key to use in sorting this event

        This is mainly used for HttpApi Events, where we need to evaluate the "$default" path (if any)
            before we evaluate any of the other paths ("/", etc), so we can make sure we don't create any
            redundant permissions. This sort places "$" before "/" or any alphanumeric characters.
        :param event: tuple of (logical_id, event_dictionary) that contains event information
        """
        logical_id, event_dict = event
        if not isinstance(event_dict, dict):
            return logical_id
        return event_dict.get("Properties", {}).get("Path", logical_id)

    def _generate_event_resources(self, lambda_function, execution_role, event_resources, lambda_alias=None):
        """Generates and returns the resources associated with this function's events.

        :param model.lambda_.LambdaFunction lambda_function: generated Lambda function
        :param iam.IAMRole execution_role: generated Lambda execution role
        :param implicit_api: Global Implicit API resource where the implicit APIs get attached to, if necessary
        :param implicit_api_stage: Global implicit API stage resource where implicit APIs get attached to, if necessary
        :param event_resources: All the event sources associated with this Lambda function
        :param model.lambda_.LambdaAlias lambda_alias: Optional Lambda Alias resource if we want to connect the
            event sources to this alias

        :returns: a list containing the function's event resources
        :rtype: list
        """
        resources = []
        if self.Events:
            for logical_id, event_dict in sorted(self.Events.items(), key=SamFunction.order_events):
                try:
                    eventsource = self.event_resolver.resolve_resource_type(event_dict).from_dict(
                        lambda_function.logical_id + logical_id, event_dict, logical_id)
                except TypeError as e:
                    raise InvalidEventException(logical_id, "{}".format(e))

                kwargs = {
                    # When Alias is provided, connect all event sources to the alias and *not* the function
                    'function': lambda_alias or lambda_function,
                    'role': execution_role,
                }

                for name, resource in event_resources[logical_id].items():
                    kwargs[name] = resource
                resources += eventsource.to_cloudformation(**kwargs)

        return resources

    def _construct_code_dict(self):
        if self.InlineCode:
            return {
                "ZipFile": self.InlineCode
            }
        elif self.CodeUri:
            return construct_s3_location_object(self.CodeUri, self.logical_id, 'CodeUri')
        else:
            raise InvalidResourceException(self.logical_id, "Either 'InlineCode' or 'CodeUri' must be set")

    def _construct_version(self, function, intrinsics_resolver):
        """Constructs a Lambda Version resource that will be auto-published when CodeUri of the function changes.
        Old versions will not be deleted without a direct reference from the CloudFormation template.

        :param model.lambda_.LambdaFunction function: Lambda function object that is being connected to a version
        :param model.intrinsics.resolver.IntrinsicsResolver intrinsics_resolver: Class that can help resolve
            references to parameters present in CodeUri. It is a common usecase to set S3Key of Code to be a
            template parameter. Need to resolve the values otherwise we will never detect a change in Code dict
        :return: Lambda function Version resource
        """
        code_dict = function.Code
        if not code_dict:
            raise ValueError("Lambda function code must be a valid non-empty dictionary")

        if not intrinsics_resolver:
            raise ValueError("intrinsics_resolver is required for versions creation")

        # Resolve references to template parameters before creating hash. This will *not* resolve all intrinsics
        # because we cannot resolve runtime values like Arn of a resource. For purposes of detecting changes, this
        # is good enough. Here is why:
        #
        # When using intrinsic functions there are two cases when has must change:
        #   - Value of the template parameter changes
        #   - (or) LogicalId of a referenced resource changes ie. !GetAtt NewResource.Arn
        #
        # Later case will already change the hash because some value in the Code dictionary changes. We handle the
        # first case by resolving references to template parameters. It is okay even if these references are
        # present inside another intrinsic such as !Join. The resolver will replace the reference with the parameter's
        # value and keep all other parts of !Join identical. This will still trigger a change in the hash.
        code_dict = intrinsics_resolver.resolve_parameter_refs(code_dict)

        # Construct the LogicalID of Lambda version by appending 10 characters of SHA of CodeUri. This is necessary
        # to trigger creation of a new version every time code location changes. Since logicalId changes, CloudFormation
        # will drop the old version and create a new one for us. We set a DeletionPolicy on the version resource to
        # prevent CloudFormation from actually deleting the underlying version resource
        #
        # SHA Collisions: For purposes of triggering a new update, we are concerned about just the difference previous
        #                 and next hashes. The chances that two subsequent hashes collide is fairly low.
        prefix = "{id}Version".format(id=self.logical_id)
        logical_id = logical_id_generator.LogicalIdGenerator(prefix, code_dict).gen()

        attributes = self.get_passthrough_resource_attributes()
        if attributes is None:
            attributes = {}
        attributes["DeletionPolicy"] = "Retain"

        lambda_version = LambdaVersion(logical_id=logical_id, attributes=attributes)
        lambda_version.FunctionName = function.get_runtime_attr('name')
        lambda_version.Description = self.VersionDescription

        return lambda_version

    def _construct_alias(self, name, function, version):
        """Constructs a Lambda Alias for the given function and pointing to the given version

        :param string name: Name of the alias
        :param model.lambda_.LambdaFunction function: Lambda function object to associate the alias with
        :param model.lambda_.LambdaVersion version: Lambda version object to associate the alias with
        :return: Lambda alias object
        :rtype model.lambda_.LambdaAlias
        """

        if not name:
            raise InvalidResourceException(self.logical_id, "Alias name is required to create an alias")

        logical_id = "{id}Alias{suffix}".format(id=function.logical_id, suffix=name)
        alias = LambdaAlias(logical_id=logical_id, attributes=self.get_passthrough_resource_attributes())
        alias.Name = name
        alias.FunctionName = function.get_runtime_attr('name')
        alias.FunctionVersion = version.get_runtime_attr("version")
        if self.ProvisionedConcurrencyConfig:
            alias.ProvisionedConcurrencyConfig = self.ProvisionedConcurrencyConfig

        return alias

    def _validate_deployment_preference_and_add_update_policy(self, deployment_preference_collection, lambda_alias,
                                                              intrinsics_resolver, mappings_resolver):
        if 'Enabled' in self.DeploymentPreference:
            self.DeploymentPreference['Enabled'] = intrinsics_resolver.resolve_parameter_refs(
                self.DeploymentPreference['Enabled'])
            if isinstance(self.DeploymentPreference['Enabled'], dict):
                raise InvalidResourceException(self.logical_id, "'Enabled' must be a boolean value")

        if 'Type' in self.DeploymentPreference:
            # resolve intrinsics and mappings for Type
            preference_type = self.DeploymentPreference['Type']
            preference_type = intrinsics_resolver.resolve_parameter_refs(preference_type)
            preference_type = mappings_resolver.resolve_parameter_refs(preference_type)
            self.DeploymentPreference['Type'] = preference_type

        if deployment_preference_collection is None:
            raise ValueError('deployment_preference_collection required for parsing the deployment preference')

        deployment_preference_collection.add(self.logical_id, self.DeploymentPreference)

        if deployment_preference_collection.get(self.logical_id).enabled:
            if self.AutoPublishAlias is None:
                raise InvalidResourceException(
                    self.logical_id,
                    "'DeploymentPreference' requires AutoPublishAlias property to be specified")
            if lambda_alias is None:
                raise ValueError('lambda_alias expected for updating it with the appropriate update policy')

            lambda_alias.set_resource_attribute("UpdatePolicy",
                                                deployment_preference_collection.update_policy(
                                                    self.logical_id).to_dict())


class SamApi(SamResourceMacro):
    """SAM rest API macro.
    """
    resource_type = 'AWS::Serverless::Api'
    property_types = {
        # Internal property set only by Implicit API plugin. If set to True, the API Event Source code will inject
        # Lambda Integration URI to the Swagger. To preserve backwards compatibility, this must be set only for
        # Implicit APIs. For Explicit APIs, customer is expected to set integration URI themselves.
        # In the future, we might rename and expose this property to customers so they can have SAM manage Explicit APIs
        # Swagger.
        '__MANAGE_SWAGGER': PropertyType(False, is_type(bool)),

        'Name': PropertyType(False, one_of(is_str(), is_type(dict))),
        'StageName': PropertyType(True, one_of(is_str(), is_type(dict))),
        'Tags': PropertyType(False, is_type(dict)),
        'DefinitionBody': PropertyType(False, is_type(dict)),
        'DefinitionUri': PropertyType(False, one_of(is_str(), is_type(dict))),
        'CacheClusterEnabled': PropertyType(False, is_type(bool)),
        'CacheClusterSize': PropertyType(False, is_str()),
        'Variables': PropertyType(False, is_type(dict)),
        'EndpointConfiguration': PropertyType(False, is_str()),
        'MethodSettings': PropertyType(False, is_type(list)),
        'BinaryMediaTypes': PropertyType(False, is_type(list)),
        'MinimumCompressionSize': PropertyType(False, is_type(int)),
        'Cors': PropertyType(False, one_of(is_str(), is_type(dict))),
        'Auth': PropertyType(False, is_type(dict)),
        'GatewayResponses': PropertyType(False, is_type(dict)),
        'AccessLogSetting': PropertyType(False, is_type(dict)),
        'CanarySetting': PropertyType(False, is_type(dict)),
        'TracingEnabled': PropertyType(False, is_type(bool)),
        'OpenApiVersion': PropertyType(False, is_str()),
        'Models': PropertyType(False, is_type(dict))
    }

    referable_properties = {
        "Stage": ApiGatewayStage.resource_type,
        "Deployment": ApiGatewayDeployment.resource_type,
    }

    def to_cloudformation(self, **kwargs):
        """Returns the API Gateway RestApi, Deployment, and Stage to which this SAM Api corresponds.

        :param dict kwargs: already-converted resources that may need to be modified when converting this \
        macro to pure CloudFormation
        :returns: a list of vanilla CloudFormation Resources, to which this Function expands
        :rtype: list
        """
        resources = []

        intrinsics_resolver = kwargs["intrinsics_resolver"]
        self.BinaryMediaTypes = intrinsics_resolver.resolve_parameter_refs(self.BinaryMediaTypes)

        api_generator = ApiGenerator(self.logical_id,
                                     self.CacheClusterEnabled,
                                     self.CacheClusterSize,
                                     self.Variables,
                                     self.depends_on,
                                     self.DefinitionBody,
                                     self.DefinitionUri,
                                     self.Name,
                                     self.StageName,
                                     tags=self.Tags,
                                     endpoint_configuration=self.EndpointConfiguration,
                                     method_settings=self.MethodSettings,
                                     binary_media=self.BinaryMediaTypes,
                                     minimum_compression_size=self.MinimumCompressionSize,
                                     cors=self.Cors,
                                     auth=self.Auth,
                                     gateway_responses=self.GatewayResponses,
                                     access_log_setting=self.AccessLogSetting,
                                     canary_setting=self.CanarySetting,
                                     tracing_enabled=self.TracingEnabled,
                                     resource_attributes=self.resource_attributes,
                                     passthrough_resource_attributes=self.get_passthrough_resource_attributes(),
                                     open_api_version=self.OpenApiVersion,
                                     models=self.Models)

        rest_api, deployment, stage, permissions = api_generator.to_cloudformation()

        resources.extend([rest_api, deployment, stage])
        resources.extend(permissions)

        return resources


class SamHttpApi(SamResourceMacro):
    """SAM rest API macro.
    """
    resource_type = 'AWS::Serverless::HttpApi'
    property_types = {
        # Internal property set only by Implicit HTTP API plugin. If set to True, the API Event Source code will
        # inject Lambda Integration URI to the OpenAPI. To preserve backwards compatibility, this must be set only for
        # Implicit APIs. For Explicit APIs, this is managed by the DefaultDefinitionBody Plugin.
        # In the future, we might rename and expose this property to customers so they can have SAM manage Explicit APIs
        # Swagger.
        '__MANAGE_SWAGGER': PropertyType(False, is_type(bool)),

        'StageName': PropertyType(False, one_of(is_str(), is_type(dict))),
        'Tags': PropertyType(False, is_type(dict)),
        'DefinitionBody': PropertyType(False, is_type(dict)),
        'DefinitionUri': PropertyType(False, one_of(is_str(), is_type(dict))),
        'StageVariables': PropertyType(False, is_type(dict)),
        'Cors': PropertyType(False, one_of(is_str(), is_type(dict))),
        'AccessLogSettings': PropertyType(False, is_type(dict)),
        'Auth': PropertyType(False, is_type(dict))
    }

    referable_properties = {
        "Stage": ApiGatewayV2Stage.resource_type,
    }

    def to_cloudformation(self, **kwargs):
        """Returns the API Gateway RestApi, Deployment, and Stage to which this SAM Api corresponds.

        :param dict kwargs: already-converted resources that may need to be modified when converting this \
        macro to pure CloudFormation
        :returns: a list of vanilla CloudFormation Resources, to which this Function expands
        :rtype: list
        """
        resources = []

        api_generator = HttpApiGenerator(self.logical_id,
                                         self.StageVariables,
                                         self.depends_on,
                                         self.DefinitionBody,
                                         self.DefinitionUri,
                                         self.StageName,
                                         tags=self.Tags,
                                         auth=self.Auth,
                                         access_log_settings=self.AccessLogSettings,
                                         resource_attributes=self.resource_attributes,
                                         passthrough_resource_attributes=self.get_passthrough_resource_attributes())

        http_api, stage = api_generator.to_cloudformation()

        resources.append(http_api)

        # Stage is now optional. Only add it if one is created.
        if stage:
            resources.append(stage)

        return resources


class SamSimpleTable(SamResourceMacro):
    """SAM simple table macro.
    """
    resource_type = 'AWS::Serverless::SimpleTable'
    property_types = {
        'PrimaryKey': PropertyType(False, dict_of(is_str(), is_str())),
        'ProvisionedThroughput': PropertyType(False, dict_of(is_str(), one_of(is_type(int), is_type(dict)))),
        'TableName': PropertyType(False, one_of(is_str(), is_type(dict))),
        'Tags': PropertyType(False, is_type(dict)),
        'SSESpecification': PropertyType(False, is_type(dict))
    }
    attribute_type_conversions = {
        'String': 'S',
        'Number': 'N',
        'Binary': 'B'
    }

    def to_cloudformation(self, **kwargs):
        dynamodb_resources = self._construct_dynamodb_table()

        return [dynamodb_resources]

    def _construct_dynamodb_table(self):
        dynamodb_table = DynamoDBTable(self.logical_id, depends_on=self.depends_on, attributes=self.resource_attributes)

        if self.PrimaryKey:
            if 'Name' not in self.PrimaryKey or 'Type' not in self.PrimaryKey:
                raise InvalidResourceException(
                    self.logical_id,
                    '\'PrimaryKey\' is missing required Property \'Name\' or \'Type\'.'
                )
            primary_key = {
                'AttributeName': self.PrimaryKey['Name'],
                'AttributeType': self._convert_attribute_type(self.PrimaryKey['Type'])
            }

        else:
            primary_key = {'AttributeName': 'id', 'AttributeType': 'S'}

        dynamodb_table.AttributeDefinitions = [primary_key]
        dynamodb_table.KeySchema = [{
            'AttributeName': primary_key['AttributeName'],
            'KeyType': 'HASH'
        }]

        if self.ProvisionedThroughput:
            dynamodb_table.ProvisionedThroughput = self.ProvisionedThroughput
        else:
            dynamodb_table.BillingMode = 'PAY_PER_REQUEST'

        if self.SSESpecification:
            dynamodb_table.SSESpecification = self.SSESpecification

        if self.TableName:
            dynamodb_table.TableName = self.TableName

        if bool(self.Tags):
            dynamodb_table.Tags = get_tag_list(self.Tags)

        return dynamodb_table

    def _convert_attribute_type(self, attribute_type):
        if attribute_type in self.attribute_type_conversions:
            return self.attribute_type_conversions[attribute_type]
        raise InvalidResourceException(self.logical_id, 'Invalid \'Type\' "{actual}".'.format(actual=attribute_type))


class SamApplication(SamResourceMacro):
    """SAM application macro.
    """

    APPLICATION_ID_KEY = 'ApplicationId'
    SEMANTIC_VERSION_KEY = 'SemanticVersion'

    resource_type = 'AWS::Serverless::Application'

    # The plugin will always insert the TemplateUrl parameter
    property_types = {
        'Location': PropertyType(True, one_of(is_str(), is_type(dict))),
        'TemplateUrl': PropertyType(False, is_str()),
        'Parameters': PropertyType(False, is_type(dict)),
        'NotificationARNs': PropertyType(False, list_of(one_of(is_str(), is_type(dict)))),
        'Tags': PropertyType(False, is_type(dict)),
        'TimeoutInMinutes': PropertyType(False, is_type(int))
    }

    def to_cloudformation(self, **kwargs):
        """Returns the stack with the proper parameters for this application
        """
        nested_stack = self._construct_nested_stack()
        return [nested_stack]

    def _construct_nested_stack(self):
        """Constructs a AWS::CloudFormation::Stack resource
        """
        nested_stack = NestedStack(self.logical_id, depends_on=self.depends_on,
                                   attributes=self.get_passthrough_resource_attributes())
        nested_stack.Parameters = self.Parameters
        nested_stack.NotificationARNs = self.NotificationARNs
        application_tags = self._get_application_tags()
        nested_stack.Tags = self._construct_tag_list(self.Tags, application_tags)
        nested_stack.TimeoutInMinutes = self.TimeoutInMinutes
        nested_stack.TemplateURL = self.TemplateUrl if self.TemplateUrl else ""

        return nested_stack

    def _get_application_tags(self):
        """Adds tags to the stack if this resource is using the serverless app repo
        """
        application_tags = {}
        if isinstance(self.Location, dict):
            if (self.APPLICATION_ID_KEY in self.Location.keys() and
                    self.Location[self.APPLICATION_ID_KEY] is not None):
                application_tags[self._SAR_APP_KEY] = self.Location[self.APPLICATION_ID_KEY]
            if (self.SEMANTIC_VERSION_KEY in self.Location.keys() and
                    self.Location[self.SEMANTIC_VERSION_KEY] is not None):
                application_tags[self._SAR_SEMVER_KEY] = self.Location[self.SEMANTIC_VERSION_KEY]
        return application_tags


class SamLayerVersion(SamResourceMacro):
    """ SAM Layer macro
    """
    resource_type = 'AWS::Serverless::LayerVersion'
    property_types = {
        'LayerName': PropertyType(False, one_of(is_str(), is_type(dict))),
        'Description': PropertyType(False, is_str()),
        'ContentUri': PropertyType(True, one_of(is_str(), is_type(dict))),
        'CompatibleRuntimes': PropertyType(False, list_of(is_str())),
        'LicenseInfo': PropertyType(False, is_str()),
        'RetentionPolicy': PropertyType(False, is_str())
    }

    RETAIN = 'Retain'
    DELETE = 'Delete'
    retention_policy_options = [RETAIN.lower(), DELETE.lower()]

    def to_cloudformation(self, **kwargs):
        """Returns the Lambda layer to which this SAM Layer corresponds.

        :param dict kwargs: already-converted resources that may need to be modified when converting this \
        macro to pure CloudFormation
        :returns: a list of vanilla CloudFormation Resources, to which this Function expands
        :rtype: list
        """
        resources = []

        # Append any CFN resources:
        intrinsics_resolver = kwargs["intrinsics_resolver"]
        resources.append(self._construct_lambda_layer(intrinsics_resolver))

        return resources

    def _construct_lambda_layer(self, intrinsics_resolver):
        """Constructs and returns the Lambda function.

        :returns: a list containing the Lambda function and execution role resources
        :rtype: list
        """
        # Resolve intrinsics if applicable:
        self.LayerName = self._resolve_string_parameter(intrinsics_resolver, self.LayerName, 'LayerName')
        self.LicenseInfo = self._resolve_string_parameter(intrinsics_resolver, self.LicenseInfo, 'LicenseInfo')
        self.Description = self._resolve_string_parameter(intrinsics_resolver, self.Description, 'Description')
        self.RetentionPolicy = self._resolve_string_parameter(intrinsics_resolver, self.RetentionPolicy,
                                                              'RetentionPolicy')

        retention_policy_value = self._get_retention_policy_value()

        attributes = self.get_passthrough_resource_attributes()
        if attributes is None:
            attributes = {}
        attributes['DeletionPolicy'] = retention_policy_value

        old_logical_id = self.logical_id
        new_logical_id = logical_id_generator.LogicalIdGenerator(old_logical_id, self.to_dict()).gen()
        self.logical_id = new_logical_id

        lambda_layer = LambdaLayerVersion(self.logical_id, depends_on=self.depends_on, attributes=attributes)

        # Changing the LayerName property: when a layer is published, it is given an Arn
        # example: arn:aws:lambda:us-west-2:123456789012:layer:MyLayer:1
        # where MyLayer is the LayerName property if it exists; otherwise, it is the
        # LogicalId of this resource. Since a LayerVersion is an immutable resource, when
        # CloudFormation updates this resource, it will ALWAYS create a new version then
        # delete the old version if the logical ids match. What this does is change the
        # logical id of every layer (so a `DeletionPolicy: Retain` can work) and set the
        # LayerName property of the layer so that the Arn will still always be the same
        # with the exception of an incrementing version number.
        if not self.LayerName:
            self.LayerName = old_logical_id

        lambda_layer.LayerName = self.LayerName
        lambda_layer.Description = self.Description
        lambda_layer.Content = construct_s3_location_object(self.ContentUri, self.logical_id, 'ContentUri')
        lambda_layer.CompatibleRuntimes = self.CompatibleRuntimes
        lambda_layer.LicenseInfo = self.LicenseInfo

        return lambda_layer

    def _get_retention_policy_value(self):
        """
        Sets the deletion policy on this resource. The default is 'Retain'.

        :return: value for the DeletionPolicy attribute.
        """

        if self.RetentionPolicy is None or self.RetentionPolicy.lower() == self.RETAIN.lower():
            return self.RETAIN
        elif self.RetentionPolicy.lower() == self.DELETE.lower():
            return self.DELETE
        elif self.RetentionPolicy.lower() not in self.retention_policy_options:
            raise InvalidResourceException(self.logical_id,
                                           "'{}' must be one of the following options: {}."
                                           .format('RetentionPolicy', [self.RETAIN, self.DELETE]))