Project operator

ProjectOperator

Bases: Operator

Project operator keeps only specified keys and optionally renames them (key-wise projection). Supports key selection and renaming with conflict detection.

Attributes:

Name	Type	Required	Default	Description
kept_keys	list[str]		-	List of keys to keep in each record
key_mapping	dict[str, str]		{}	Dictionary mapping old key names to new key names

Source code in blue/operators/project_operator.py

class ProjectOperator(Operator):
    """
    Project operator keeps only specified keys and optionally renames them (key-wise projection).
    Supports key selection and renaming with conflict detection.

    Attributes:
    ----------
    | Name        | Type         | Required | Default | Description                                |
    |------------|-------------|---------|--------|--------------------------------------------|
    | `kept_keys`  | list[str]   | :fontawesome-solid-circle-check: {.green-check}    | -      | List of keys to keep in each record       |
    | `key_mapping`| dict[str, str]|    | {}     | Dictionary mapping old key names to new key names |
    """

    PROPERTIES = {}

    name = "project"
    description = "Given an input data, return only a set of attributes for each data element (key-wise)"
    default_attributes = {
        "kept_keys": {"type": "list[str]", "description": "List of keys to keep in each record", "required": True},
        "key_mapping": {"type": "dict[str, str]", "description": "Dictionary mapping old key names to new key names", "required": False, "default": {}},
    }

    def __init__(self, description: str = None, properties: Dict[str, Any] = None):
        super().__init__(
            self.name,
            function=project_operator_function,
            description=description or self.description,
            properties=properties,
            validator=project_operator_validator,
            explainer=project_operator_explainer,
        )

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

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

project_operator_explainer(output, input_data, attributes)

Generate explanation for project 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/project_operator.py

def project_operator_explainer(output: Any, input_data: List[List[Dict[str, Any]]], attributes: Dict[str, Any]) -> Dict[str, Any]:
    """Generate explanation for project 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)

project_operator_function(input_data, attributes, properties=None)

Project records to keep only specified keys and optionally rename them (key-wise projection).

Parameters:

Name	Type	Description	Default
input_data	List[List[Dict[str, Any]]]	List of JSON arrays (List[List[Dict[str, Any]]]), uses the first data source for projection.	required
attributes	Dict[str, Any]	Dictionary containing projection parameters including kept_keys and key_mapping.	required
properties	Dict[str, Any]	Optional properties dictionary. Defaults to None.	None

Returns:

Type	Description
List[List[Dict[str, Any]]]	List containing projected records with only the specified keys.

Source code in blue/operators/project_operator.py

def project_operator_function(input_data: List[List[Dict[str, Any]]], attributes: Dict[str, Any], properties: Dict[str, Any] = None) -> List[List[Dict[str, Any]]]:
    """Project records to keep only specified keys and optionally rename them (key-wise projection).

    Parameters:
        input_data: List of JSON arrays (List[List[Dict[str, Any]]]), uses the first data source for projection.
        attributes: Dictionary containing projection parameters including kept_keys and key_mapping.
        properties: Optional properties dictionary. Defaults to None.

    Returns:
        List containing projected records with only the specified keys.
    """
    # Extract attributes
    kept_keys = attributes.get('kept_keys', [])
    key_mapping = attributes.get('key_mapping', {})

    # Validate input
    if not input_data or not input_data[0]:
        return []

    if not kept_keys:
        return []

    data = input_data[0]  # Use first data source

    # Project records
    result = []
    for record in data:
        projected_record = {}
        for key in kept_keys:
            if key in record:
                # Apply key mapping if specified
                output_key = key_mapping.get(key, key)
                projected_record[output_key] = record[key]
        result.append(projected_record)
    return [result]

project_operator_validator(input_data, attributes, properties=None)

Validate project 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/project_operator.py

def project_operator_validator(input_data: List[List[Dict[str, Any]]], attributes: Dict[str, Any], properties: Dict[str, Any] = None) -> bool:
    """Validate project 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 (types already checked by default validator)
    kept_keys = attributes.get('kept_keys', [])
    key_mapping = attributes.get('key_mapping', {})

    # Validate that all keys in key_mapping are in kept_keys
    for mapped_key in key_mapping.keys():
        if mapped_key not in kept_keys:
            return False

    try:
        # Validate key mapping conflicts
        _validate_key_mapping_conflicts(kept_keys, key_mapping)
    except ValueError:
        return False

    return True

Last update: 2025-10-08