def _inject_notification_configuration()

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

• 32 lines of code
• 10 McCabe index (conditional complexity)

    def _inject_notification_configuration(self, function, bucket, bucket_id):  # type: ignore[no-untyped-def]
        base_event_mapping = {"Function": function.get_runtime_attr("arn")}

        if self.Filter is not None:
            base_event_mapping["Filter"] = self.Filter

        event_types = self.Events
        if isinstance(self.Events, str):
            event_types = [self.Events]

        event_mappings = []
        for event_type in event_types:
            lambda_event = copy.deepcopy(base_event_mapping)
            lambda_event["Event"] = event_type
            if CONDITION in function.resource_attributes:
                lambda_event = make_conditional(function.resource_attributes[CONDITION], lambda_event)
            event_mappings.append(lambda_event)

        properties = bucket.get("Properties", {})
        sam_expect(properties, bucket_id, "").to_be_a_map("Properties should be a map.")
        bucket["Properties"] = properties

        notification_config = properties.get("NotificationConfiguration", None)
        if notification_config is None:
            notification_config = {}
            properties["NotificationConfiguration"] = notification_config

        sam_expect(notification_config, bucket_id, "NotificationConfiguration").to_be_a_map()

        lambda_notifications = notification_config.get("LambdaConfigurations", None)
        if lambda_notifications is None:
            lambda_notifications = []
            notification_config["LambdaConfigurations"] = lambda_notifications

        if not isinstance(lambda_notifications, list):
            raise InvalidResourceException(bucket_id, "Invalid type for LambdaConfigurations. Must be a list.")

        for event_mapping in event_mappings:
            if event_mapping not in lambda_notifications:
                lambda_notifications.append(event_mapping)
        return bucket