SH-Deployer/apps/SA-ITOA/lib/itsi/content_packs/writers.py

# Copyright (C) 2005-2025 Splunk Inc. All Rights Reserved.
import sys

from itsi.content_packs.constants import ContentType, CONTENT_TYPE_WRITE_PRIORITY
from itsi.content_packs.itoa import (
    get_itoa_object_class,
    get_itoa_object_id,
    get_itoa_object_title
)
from itsi.content_packs.journal import EntryType
from ITOA.setup_logging import InstrumentCall


class ItoaContentWriter(object):
    """
    Handles the saving of content objects into ITSI.
    """

    def __init__(self, logger, session_key, transaction_id=None):
        """
        :param logger: a logger instance
        :type logger: Logger

        :param session_key: the session key
        :type session_key: str

        :param transaction_id: transaction info for tracking for debugging
        :type transaction_id: basestring
        """
        self.logger = logger
        self.session_key = session_key
        self._instrumentation = InstrumentCall(logger)
        self.transaction_id = transaction_id

    def write(self, content_objects, journal):
        """
        Writes the given content objects to storage.

        :param content_objects: the content objects data
        :type content_objects: dict

        :param journal: a journal instance
        :type journal: TransactionJournal

        :return: a dict of content objects data
        :rtype: dict
        """
        with self._instrumentation.track("ItoaContentWriter.write", transaction_id=self.transaction_id):
            failed_parent_itsi_object_priority_order = sys.maxsize
            failed_parent_itsi_object_type = ""
            results = {}

            content_types_with_priority = sorted(((content_type, CONTENT_TYPE_WRITE_PRIORITY[content_type]) for content_type in content_objects.keys()), key=lambda x: x[1])

            for content_type, priority_order in content_types_with_priority:

                objects = content_objects[content_type]
                if not objects:
                    continue

                if (failed_parent_itsi_object_type
                        and failed_parent_itsi_object_type not in (ContentType.GLASS_TABLE_ICON, ContentType.GLASS_TABLE_IMAGE)
                        and failed_parent_itsi_object_priority_order < priority_order):
                    error_message = f'tid={self.transaction_id} Skipping installation of {content_type} due to a failure while writing the parent ITSI Object Type: "{failed_parent_itsi_object_type}"'
                    self.logger.error(error_message)
                    journal.failure({
                        'error_message': error_message,
                        'type': EntryType.SKIPPING_INSTALLATION_WRITING + ' ' + failed_parent_itsi_object_type,
                        'content_type': content_type,
                        'ids': list({get_itoa_object_id(content_type, obj) for obj in objects}),
                        'titles': list({get_itoa_object_title(content_type, obj) for obj in objects})
                    })
                else:
                    object_ids, failed_parent_itsi_object_type, failed_parent_itsi_object_priority_order = self.write_objects(content_type, objects, priority_order, journal=journal)

                    objects_data = self.construct_objects_data(content_type, object_ids, objects)
                    if objects_data:
                        results[content_type] = objects_data

            return results

    def write_objects(self, content_type, objects, priority_order, journal):
        """
        Writes the given objects for a content type to storage.

        :param content_type: the content type
        :type content_type: str

        :param objects: a list of objects data
        :type objects: list

        :param priority_order: priority order of ITSI Object
        :type priority_order: int

        :param journal: a journal instance
        :type journal: TransactionJournal

        :return: a tuple that will contain following items:
                 1. list of successfully written object ids
                 2. Name of ITSI Object Type that indicates that installation is skipped for dependent ITSI Objects Type
                 3. Priority Order of ITSI Object Type
        :rtype: (list, string, int)
        """
        with self._instrumentation.track(f"ItoaContentWriter.write_objects.{content_type}", transaction_id=self.transaction_id):
            itoa_object_class = get_itoa_object_class(content_type)
            if not itoa_object_class:
                error_message = f'tid={self.transaction_id} Unexpected writing of unsupported content_type="{content_type}"'
                self.logger.error(error_message)

                journal.failure({
                    'error_message': error_message,
                    'type': EntryType.INVALID_CONTENT_TYPE,
                    'content_type': content_type,
                    'ids': list({get_itoa_object_id(content_type, obj) for obj in objects}),
                    'titles': list({get_itoa_object_title(content_type, obj) for obj in objects})
                })

                return [], content_type, priority_order

            handler = itoa_object_class(self.session_key, 'nobody')
            handler.skip_service_template_update = True

            try:
                object_ids = handler.save_batch(
                    'nobody',
                    objects,
                    validate_names=False,
                    transaction_id=self.transaction_id
                )
                self.logger.info(f'tid={self.transaction_id} Successfully created objects for content_type="{content_type}"')
            except Exception as ex:
                self.logger.error(f'tid={self.transaction_id} Failed to create objects for content_type="{content_type}"')
                self.logger.exception(f'tid={self.transaction_id} {ex}')

                journal.failure({
                    'error_message': str(ex),
                    'type': EntryType.ERROR_OBJECTS_SAVE,
                    'content_type': content_type,
                    'ids': list({get_itoa_object_id(content_type, obj) for obj in objects}),
                    'titles': list({get_itoa_object_title(content_type, obj) for obj in objects})
                })

                return [], content_type, priority_order

            return object_ids, "", sys.maxsize

    def construct_objects_data(self, content_type, object_ids, objects):
        """
        Constructs the objects data for the given list of object ids.

        :param content_type: the content type
        :type content_type: str

        :param object_ids: a list of object ids
        :type object_ids: list

        :param objects: a list of object data
        :type objects: list

        :return: a list of object data
        :rtype: list
        """
        lookup = { get_itoa_object_id(content_type, obj): obj for obj in objects }

        objects_data = []

        for object_id in object_ids:
            if object_id not in lookup:
                continue

            data = lookup[object_id]

            objects_data.append({
                'id': object_id,
                'title': get_itoa_object_title(content_type, data)
            })

        return objects_data