Select operator

SelectOperator

Bases: Operator

Select operator filters records based on a single condition (record-wise filtering). Supports basic comparison operators with type-aware comparison logic.

Attributes:

Name	Type	Required	Default	Description
operand_key	str		-	The key to check in each record
operand	str		-	Comparison operator: =, !=, >, >=, <, <=
operand_val	Any		-	Value to compare with
approximate_match	bool		False	Use epsilon tolerance for numeric comparison
eps	float		1e-9	Epsilon tolerance for approximate numeric comparison

Source code in blue/operators/select_operator.py

class SelectOperator(Operator):
    """
    Select operator filters records based on a single condition (record-wise filtering).
    Supports basic comparison operators with type-aware comparison logic.

    Attributes:
    ----------
    | Name               | Type  | Required | Default | Description                                           |
    |-------------------|-------|---------|--------|-------------------------------------------------------|
    | `operand_key`        | str   | :fontawesome-solid-circle-check: {.green-check}    | -      | The key to check in each record                       |
    | `operand`            | str   | :fontawesome-solid-circle-check: {.green-check}    | -      | Comparison operator: =, !=, >, >=, <, <=             |
    | `operand_val`        | Any   | :fontawesome-solid-circle-check: {.green-check}    | -      | Value to compare with                                  |
    | `approximate_match`  | bool  |    | False  | Use epsilon tolerance for numeric comparison          |
    | `eps`                | float |    | 1e-9   | Epsilon tolerance for approximate numeric comparison  |

    """

    PROPERTIES = {}

    name = "select"
    description = "Given an input data, filter data elements based on a specified condition by type-aware comparison (record-wise)"
    default_attributes = {
        "operand_key": {"type": "str", "description": "The key to check in each record", "required": True},
        "operand": {"type": "str", "description": "Comparison operator: =, !=, >, >=, <, <=", "required": True},
        "operand_val": {"type": "Any", "description": "Value to compare with", "required": True},
        "approximate_match": {"type": "bool", "description": "Use epsilon tolerance for numeric comparison", "required": False, "default": False},
        "eps": {"type": "float", "description": "Epsilon tolerance for approximate numeric comparison", "required": False, "default": 1e-9},
    }

    def __init__(self, description: str = None, properties: Dict[str, Any] = None):
        super().__init__(
            self.name,
            function=select_operator_function,
            description=description or self.description,
            properties=properties,
            validator=select_operator_validator,
            explainer=select_operator_explainer,
        )

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

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

select_operator_explainer(output, input_data, attributes)

Generate explanation for select 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/select_operator.py

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

select_operator_function(input_data, attributes, properties=None)

Filter records based on a single condition (record-wise filtering).

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 filtering.	required
attributes	Dict[str, Any]	Dictionary containing filtering parameters including operand_key, operand, operand_val, approximate_match, and eps.	required
properties	Dict[str, Any]	Optional properties dictionary. Defaults to None.	None

Returns:

Type	Description
List[List[Dict[str, Any]]]	List containing filtered records from the first data source.

Source code in blue/operators/select_operator.py

def select_operator_function(input_data: List[List[Dict[str, Any]]], attributes: Dict[str, Any], properties: Dict[str, Any] = None) -> List[List[Dict[str, Any]]]:
    """Filter records based on a single condition (record-wise filtering).

    Parameters:
        input_data: List of JSON arrays (List[List[Dict[str, Any]]]), uses the first data source for filtering.
        attributes: Dictionary containing filtering parameters including operand_key, operand, operand_val, approximate_match, and eps.
        properties: Optional properties dictionary. Defaults to None.

    Returns:
        List containing filtered records from the first data source.
    """
    # Extract attributes
    operand_key = attributes.get('operand_key')
    operand = attributes.get('operand')
    operand_val = attributes.get('operand_val')
    approximate_match = attributes.get('approximate_match', False)
    eps = attributes.get('eps', 1e-9)

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

    data = input_data[0]  # Use first data source

    # Filter records based on condition
    result = []
    for record in data:
        if operand_key not in record:
            continue
        record_value = record[operand_key]
        if _evaluate_condition(record_value, operand, operand_val, approximate_match, eps):
            result.append(record)
    return [result]

select_operator_validator(input_data, attributes, properties=None)

Validate select 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/select_operator.py

def select_operator_validator(input_data: List[List[Dict[str, Any]]], attributes: Dict[str, Any], properties: Dict[str, Any] = None) -> bool:
    """Validate select 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)
    operand = attributes.get('operand')
    if operand and operand not in ['=', '!=', '>', '>=', '<', '<=']:
        return False

    eps = attributes.get('eps', 1e-9)
    if eps < 0:
        return False

    return True

Last update: 2025-10-08