def to_cloudformation()

in samtranslator/model/eventsources/push.py [0:0]

• 21 lines of code
• 9 McCabe index (conditional complexity)

    def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def]
        """Returns the Lambda Permission resource allowing S3 to invoke the function this event source triggers.

        :param dict kwargs: S3 bucket resource
        :returns: a list of vanilla CloudFormation Resources, to which this S3 event expands
        :rtype: list
        """
        function = kwargs.get("function")

        if not function:
            raise TypeError("Missing required keyword argument: function")

        if "bucket" not in kwargs or kwargs["bucket"] is None:
            raise TypeError("Missing required keyword argument: bucket")

        if "bucket_id" not in kwargs or kwargs["bucket_id"] is None:
            raise TypeError("Missing required keyword argument: bucket_id")

        bucket = kwargs["bucket"]
        bucket_id = kwargs["bucket_id"]

        resources = []

        source_account = ref("AWS::AccountId")
        permission = self._construct_permission(function, source_account=source_account)  # type: ignore[no-untyped-call]
        if CONDITION in permission.resource_attributes:
            self._depend_on_lambda_permissions_using_tag(bucket, bucket_id, permission)
        else:
            self._depend_on_lambda_permissions(bucket, permission)  # type: ignore[no-untyped-call]
        resources.append(permission)

        # NOTE: `bucket` here is a dictionary representing the S3 Bucket resource in your SAM template. If there are
        # multiple S3 Events attached to the same bucket, we will update the Bucket resource with notification
        # configuration for each event. This is the reason why we continue to use existing bucket dict and append onto
        # it.
        #
        # NOTE: There is some fragile logic here where we will append multiple resources to output
        #   SAM template but de-dupe them when merging into output CFN template. This is scary because the order of
        #   merging is literally "last one wins", which works fine because we linearly loop through the template once.
        #   The de-dupe happens inside `samtranslator.translator.Translator.translate` method when merging results of
        #   to_cloudformation() to output template.
        self._inject_notification_configuration(function, bucket, bucket_id)  # type: ignore[no-untyped-call]
        resources.append(S3Bucket.from_dict(bucket_id, bucket))

        return resources