Insert operator

InsertOperator

Bases: Operator

Insert operator inserts records into the first data source at specified positions. Records come from attributes or second data group.

Attributes:

Name	Type	Required	Default	Description
insert_records	list[dict]		[]	List of records to insert into the first data source (optional if second data group provided)
insert_idx	Union[int, list[int]]		-1	Position(s) to insert records (-1 for append, 0+ for specific position). If list, each element corresponds to position for each record.

Source code in blue/operators/insert_operator.py

class InsertOperator(Operator):
    """
    Insert operator inserts records into the first data source at specified positions.
    Records come from attributes or second data group.

    Attributes:
    ----------
    | Name           | Type                 | Required | Default | Description                                                                                           |
    |----------------|--------------------|----------|---------|-------------------------------------------------------------------------------------------------------|
    | `insert_records` | list[dict]          |     | []      | List of records to insert into the first data source (optional if second data group provided)       |
    | `insert_idx`     | Union[int, list[int]] |     | -1      | Position(s) to insert records (-1 for append, 0+ for specific position). If list, each element corresponds to position for each record.                         |

    """

    PROPERTIES = {}

    name = "insert"
    description = "Given input data, insert records by position(s)"
    default_attributes = {
        "insert_records": {
            "type": "list[dict]",
            "description": "List of records to insert into the first data source (optional if second data group provided)",
            "required": False,
            "default": [],
        },
        "insert_idx": {
            "type": "Union[int, list[int]]",
            "description": "Position(s) to insert records (-1 for append, 0+ for specific position). If list, each element corresponds to position for each record",
            "required": False,
            "default": -1,
        },
    }

    def __init__(self, description: str = None, properties: Dict[str, Any] = None):
        super().__init__(
            self.name,
            function=insert_operator_function,
            description=description or self.description,
            properties=properties,
            validator=insert_operator_validator,
            explainer=insert_operator_explainer,
        )

    def _initialize_properties(self):
        super()._initialize_properties()

        # attribute definitions
        self.properties["attributes"] = self.default_attributes

insert_operator_explainer(output, input_data, attributes)

Generate explanation for insert operator execution.

Parameters:

Name	Type	Description	Default
output	Any	The output result from the operator execution.	required
input_data	List[List[Dict[str, Any]]]	The input data that was processed.	required
attributes	Dict[str, Any]	The attributes used for the operation.	required

Returns:

Type	Description
Dict[str, Any]	Dictionary containing explanation of the operation.

Source code in blue/operators/insert_operator.py

def insert_operator_explainer(output: Any, input_data: List[List[Dict[str, Any]]], attributes: Dict[str, Any]) -> Dict[str, Any]:
    """Generate explanation for insert operator execution.

    Parameters:
        output: The output result from the operator execution.
        input_data: The input data that was processed.
        attributes: The attributes used for the operation.

    Returns:
        Dictionary containing explanation of the operation.
    """
    return default_operator_explainer(output, input_data, attributes)

insert_operator_function(input_data, attributes, properties=None)

Insert records into the first data source at specified positions.

Parameters:

Name	Type	Description	Default
input_data	List[List[Dict[str, Any]]]	List of JSON arrays (List[List[Dict[str, Any]]]), uses first data source as base and optionally second data source for records to insert.	required
attributes	Dict[str, Any]	Dictionary containing insert parameters including insert_records and insert_idx.	required
properties	Dict[str, Any]	Optional properties dictionary. Defaults to None.	None

Returns:

Type	Description
List[List[Dict[str, Any]]]	List containing the first data source with records inserted at specified positions.

Source code in blue/operators/insert_operator.py

def insert_operator_function(input_data: List[List[Dict[str, Any]]], attributes: Dict[str, Any], properties: Dict[str, Any] = None) -> List[List[Dict[str, Any]]]:
    """Insert records into the first data source at specified positions.

    Parameters:
        input_data: List of JSON arrays (List[List[Dict[str, Any]]]), uses first data source as base and optionally second data source for records to insert.
        attributes: Dictionary containing insert parameters including insert_records and insert_idx.
        properties: Optional properties dictionary. Defaults to None.

    Returns:
        List containing the first data source with records inserted at specified positions.
    """
    insert_idx = attributes.get('insert_idx', -1)
    if not input_data or not input_data[0]:
        return []

    # Get insert_records from second data group or attributes
    if len(input_data) >= 2 and input_data[1]:
        insert_records = input_data[1]
    else:
        insert_records = attributes.get('insert_records', [])

    if not insert_records:
        return [input_data[0].copy()]

    base_data = input_data[0]
    base_len = len(base_data)

    # Handle single position
    if isinstance(insert_idx, int):
        pos = insert_idx
        if pos < 0 or pos > base_len:
            return [base_data + insert_records]
        else:
            return [base_data[:pos] + insert_records + base_data[pos:]]

    # Handle multiple positions
    else:
        insert_positions = insert_idx.copy()
        while len(insert_positions) < len(insert_records):
            insert_positions.append(-1)

        return [_build_result_with_multiple_inserts(base_data, insert_records, insert_positions)]

insert_operator_validator(input_data, attributes, properties=None)

Validate insert operator attributes.

Parameters:

Name	Type	Description	Default
input_data	List[List[Dict[str, Any]]]	List of JSON arrays (List[List[Dict[str, Any]]]) to validate.	required
attributes	Dict[str, Any]	Dictionary containing operator attributes to validate.	required
properties	Dict[str, Any]	Optional properties dictionary. Defaults to None.	None

Returns:

Type	Description
bool	True if attributes are valid, False otherwise.

Source code in blue/operators/insert_operator.py

def insert_operator_validator(input_data: List[List[Dict[str, Any]]], attributes: Dict[str, Any], properties: Dict[str, Any] = None) -> bool:
    """Validate insert operator attributes.

    Parameters:
        input_data: List of JSON arrays (List[List[Dict[str, Any]]]) to validate.
        attributes: Dictionary containing operator attributes to validate.
        properties: Optional properties dictionary. Defaults to None.

    Returns:
        True if attributes are valid, False otherwise.
    """
    try:
        if not default_operator_validator(input_data, attributes, properties):
            return False
    except Exception:
        return False

    # Business logic validation
    insert_idx = attributes.get('insert_idx', -1)

    # Validate insert_idx values are not less than -1
    if isinstance(insert_idx, int):
        if insert_idx < -1:
            return False
    elif isinstance(insert_idx, list):
        for idx in insert_idx:
            if idx < -1:
                return False

    return True

Last update: 2025-10-08