def to_cloudformation()

in samtranslator/model/sam_resources.py [0:0]
66 lines of code
12 McCabe index (conditional complexity)

    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)
        conditions = kwargs.get("conditions", {})

        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
        alias_name = ""
        if self.AutoPublishAlias:
            alias_name = self._get_resolved_alias_name("AutoPublishAlias", self.AutoPublishAlias, intrinsics_resolver)
            code_sha256 = None
            if self.AutoPublishCodeSha256:
                code_sha256 = intrinsics_resolver.resolve_parameter_refs(self.AutoPublishCodeSha256)
                if not isinstance(code_sha256, str):
                    raise InvalidResourceException(
                        self.logical_id,
                        "AutoPublishCodeSha256 must be a string",
                    )
            lambda_version = self._construct_version(
                lambda_function, intrinsics_resolver=intrinsics_resolver, code_sha256=code_sha256
            )
            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,
            )
        event_invoke_policies = []
        if self.EventInvokeConfig:
            function_name = lambda_function.logical_id
            event_invoke_resources, event_invoke_policies = self._construct_event_invoke_config(
                function_name, alias_name, lambda_alias, intrinsics_resolver, conditions
            )
            resources.extend(event_invoke_resources)

        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, event_invoke_policies)
            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"],
                intrinsics_resolver,
                lambda_alias=lambda_alias,
            )
        except InvalidEventException as e:
            raise InvalidResourceException(self.logical_id, e.message)

        return resources