def to_cloudformation()

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

    def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def] # noqa: PLR0915
        """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: List[Any] = []
        intrinsics_resolver: IntrinsicsResolver = kwargs["intrinsics_resolver"]
        resource_resolver: ResourceResolver = kwargs["resource_resolver"]
        mappings_resolver: Optional[IntrinsicsResolver] = kwargs.get("mappings_resolver")
        conditions = kwargs.get("conditions", {})
        feature_toggle = kwargs.get("feature_toggle")

        if self.DeadLetterQueue:
            self._validate_dlq(self.DeadLetterQueue)

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

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

        lambda_alias: Optional[LambdaAlias] = 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 doesn't create a new version if the code in the unpublished version is the same as the
                # previous published version. To address situations where users modify only the 'CodeUri' content,
                # CloudFormation might not detect any changes in the Lambda function within the template, leading
                # to deployment issues. To resolve this, we'll append codesha256 value to the description.
                description = intrinsics_resolver.resolve_parameter_refs(self.Description)
                if not description or isinstance(description, str):
                    lambda_function.Description = f"{description} {code_sha256}" if description else code_sha256
                else:
                    lambda_function.Description = {"Fn::Join": [" ", [description, code_sha256]]}
            lambda_version = self._construct_version(
                lambda_function,
                intrinsics_resolver=intrinsics_resolver,
                resource_resolver=resource_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.FunctionUrlConfig:
            lambda_url = self._construct_function_url(lambda_function, lambda_alias, self.FunctionUrlConfig)
            resources.append(lambda_url)
            url_permission = self._construct_url_permission(lambda_function, lambda_alias, self.FunctionUrlConfig)
            if url_permission:
                resources.append(url_permission)

        self._validate_deployment_preference_and_add_update_policy(
            kwargs.get("deployment_preference_collection"),
            lambda_alias,
            intrinsics_resolver,
            cast(IntrinsicsResolver, mappings_resolver),  # TODO: better handle mappings_resolver's Optional
            self.get_passthrough_resource_attributes(),
            feature_toggle,
        )

        event_invoke_policies: List[Dict[str, Any]] = []
        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, self.EventInvokeConfig
            )
            resources.extend(event_invoke_resources)

        managed_policy_map = kwargs.get("managed_policy_map", {})
        get_managed_policy_map = kwargs.get("get_managed_policy_map")

        execution_role = None
        if lambda_function.Role is None:
            execution_role = self._construct_role(
                managed_policy_map,
                event_invoke_policies,
                intrinsics_resolver,
                get_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"],
                intrinsics_resolver,
                lambda_alias=lambda_alias,
                original_template=kwargs.get("original_template"),
            )
        except InvalidEventException as e:
            raise InvalidResourceException(self.logical_id, e.message) from e

        self.propagate_tags(resources, self.Tags, self.PropagateTags)

        return resources