Method add

zappa/utilities.py:439–485

(self, function_arn: str)

        return function_arn.split(":")[-1]

    def add(self, function_arn: str) -> None:
        # Add Lambda permission
        try:
            self._lambda.add_permission(
                FunctionName=function_arn,
                StatementId=f"s3-{self.bucket_name}",
                Action="lambda:InvokeFunction",
                Principal="s3.amazonaws.com",
                SourceArn=self.arn,
            )
        except botocore.exceptions.ClientError as e:
            if e.response["Error"]["Code"] != "ResourceConflictException":
                LOG.exception("Unable to add Lambda permission for S3")

        # Configure bucket notification
        try:
            # Get existing configuration
            try:
                response = self._s3.get_bucket_notification_configuration(Bucket=self.bucket_name)
                config = response
            except botocore.exceptions.ClientError:
                config = {}

            # Add new configuration
            lambda_configs = config.get("LambdaFunctionConfigurations", [])
            new_config: Dict[str, Any] = {
                "Id": self._make_notification_id(function_arn),
                "LambdaFunctionArn": function_arn,
                "Events": self.events,
            }
            if self.prefix:
                new_config["Filter"] = {"Key": {"FilterRules": [{"Name": "prefix", "Value": self.prefix}]}}
            if self.suffix:
                if "Filter" not in new_config:
                    new_config["Filter"] = {"Key": {"FilterRules": []}}
                new_config["Filter"]["Key"]["FilterRules"].append({"Name": "suffix", "Value": self.suffix})

            lambda_configs.append(new_config)
            config["LambdaFunctionConfigurations"] = lambda_configs

            # Remove ResponseMetadata if present
            config.pop("ResponseMetadata", None)

            self._s3.put_bucket_notification_configuration(Bucket=self.bucket_name, NotificationConfiguration=config)
            LOG.debug("Added S3 event source")
        except Exception:
            LOG.exception("Unable to add S3 event source")

    def remove(self, function_arn: str) -> bool:
        try: