src/backend/base/langflow/components/processing/merge_data.py

from loguru import logger

from langflow.custom import Component
from langflow.io import DataInput, Output
from langflow.schema import Data


class MergeDataComponent(Component):
    """MergeDataComponent is responsible for combining multiple Data objects into a unified list of Data objects.

    It ensures that all keys across the input Data objects are present in each merged Data object.
    Missing keys are filled with empty strings to maintain consistency.
    """

    display_name = "Merge Data"
    description = (
        "Combines multiple Data objects into a unified list, ensuring all keys are present in each Data object."
    )
    icon = "merge"

    inputs = [
        DataInput(
            name="data_inputs",
            display_name="Data Inputs",
            is_list=True,
            info="A list of Data inputs objects to be merged.",
        ),
    ]

    outputs = [
        Output(
            display_name="Merged Data",
            name="merged_data",
            method="merge_data",
        ),
    ]

    def merge_data(self) -> list[Data]:
        """Merges multiple Data objects into a single list of Data objects.

        Ensures that all keys from the input Data objects are present in each merged Data object.
        Missing keys are filled with empty strings.

        Returns:
            List[Data]: A list of merged Data objects with consistent keys.
        """
        logger.info("Initiating the data merging process.")

        data_inputs: list[Data] = self.data_inputs
        logger.debug(f"Received {len(data_inputs)} data input(s) for merging.")

        if not data_inputs:
            logger.warning("No data inputs provided. Returning an empty list.")
            return []

        # Collect all unique keys from all Data objects
        all_keys: set[str] = set()
        for idx, data_input in enumerate(data_inputs):
            if not isinstance(data_input, Data):
                error_message = f"Data input at index {idx} is not of type Data."
                logger.error(error_message)
                type_error_message = (
                    f"All items in data_inputs must be of type Data. Item at index {idx} is {type(data_input)}"
                )
                raise TypeError(type_error_message)
            all_keys.update(data_input.data.keys())
        logger.debug(f"Collected {len(all_keys)} unique key(s) from input data.")

        try:
            # Create new list of Data objects with missing keys filled with empty strings
            merged_data_list = []
            for idx, data_input in enumerate(data_inputs):
                merged_data_dict = {}

                for key in all_keys:
                    # Use the existing value if the key exists, otherwise use an empty string
                    value = data_input.data.get(key, "")
                    if key not in data_input.data:
                        log_message = f"Key '{key}' missing in data input at index {idx}. " "Assigning empty string."
                        logger.debug(log_message)
                    merged_data_dict[key] = value

                merged_data = Data(
                    text_key=data_input.text_key, data=merged_data_dict, default_value=data_input.default_value
                )
                merged_data_list.append(merged_data)
                logger.debug("Merged Data object created for input at index: " + str(idx))

        except Exception:
            logger.exception("An error occurred during the data merging process.")
            raise

        logger.info("Data merging process completed successfully.")
        return merged_data_list