Evaluation Module¶

Safe expression evaluation for conditions.

ConditionEvaluator¶

Safely evaluates condition expressions against flow context.

ConditionEvaluator ¶

ConditionEvaluator()

Safely evaluates condition expressions.

Uses AST validation to ensure expressions don't contain dangerous constructs before evaluation.

Example

evaluator = ConditionEvaluator()
context = FlowContext()
context.set("user", {"active": True})

result = evaluator.evaluate(
    "context.data.user.active == True",
    context
)
print(result)  # True

Initialize evaluator with AST validator.

evaluate ¶

evaluate(condition: str, context: FlowContext) -> bool

Evaluate a condition expression.

Parameters:

Name	Type	Description	Default
`condition`	`str`	Python expression string	required
`context`	`FlowContext`	Current flow context	required

Returns:

Type	Description
`bool`	Boolean result of evaluation

Raises:

Type	Description
`ConditionEvaluationError`	If condition is unsafe or invalid

is_safe ¶

is_safe(condition: str) -> bool

Check if a condition is safe to evaluate.

Parameters:

Name	Type	Description	Default
`condition`	`str`	Python expression string	required

Returns:

Type	Description
`bool`	True if condition passes safety validation

validate ¶

validate(condition: str) -> list[str]

Validate a condition and return any errors.

Parameters:

Name	Type	Description	Default
`condition`	`str`	Python expression string	required

Returns:

Type	Description
`list[str]`	List of validation errors (empty if valid)

SafeASTValidator¶

Validates Python AST for safe expressions only.

SafeASTValidator ¶

SafeASTValidator()

Bases: NodeVisitor

Validates that an AST only contains safe nodes.

Prevents code injection by rejecting: - Function calls - Imports - Attribute assignment - Lambda/comprehensions - Any assignment operations

Example

validator = SafeASTValidator()

# Safe expression
tree = ast.parse("x > 5 and y == 'test'", mode='eval')
assert validator.validate(tree)

# Unsafe expression (function call)
tree = ast.parse("len(items) > 0", mode='eval')
assert not validator.validate(tree)
print(validator.errors)  # ["Disallowed node type: Call"]

Initialize validator with empty error list.

validate ¶

validate(node: AST) -> bool

Validate entire AST tree.

Parameters:

Name	Type	Description	Default
`node`	`AST`	Root AST node to validate	required

Returns:

Type	Description
`bool`	True if all nodes are safe, False otherwise

get_errors ¶

get_errors() -> list[str]

Get list of validation errors.

Returns:

Type	Description
`list[str]`	List of error messages

Allowed Expression Constructs¶

Comparison Operators¶

Operator	Example
`==`	`context.data.status == 'active'`
`!=`	`context.data.status != 'deleted'`
`<`	`context.data.count < 100`
`<=`	`context.data.count <= 100`
`>`	`context.data.count > 0`
`>=`	`context.data.count >= 10`

Boolean Operators¶

Operator	Example
`and`	`a == 1 and b == 2`
`or`	`status == 'a' or status == 'b'`
`not`	`not is_disabled`

Identity and Membership¶

Operator	Example
`is`	`value is None`
`is not`	`value is not None`
`in`	`status in ['active', 'pending']`
`not in`	`role not in ['guest']`

Attribute and Subscript Access¶

context.data.user.name          # Attribute access
context.data.items[0]           # Index access
context.data["key"]             # Key access

Literals¶

True, False, None               # Boolean and None
42, 3.14, -1                    # Numbers
"string", 'string'              # Strings
[1, 2, 3]                       # Lists
("a", "b")                      # Tuples
{"key": "value"}                # Dicts

Disallowed Constructs¶

These are blocked for security:

Construct	Example	Reason
Function calls	`len(x)`	Code execution
Imports	`import os`	Module access
Assignments	`x = 1`	State mutation
Lambda	`lambda: x`	Code execution
Comprehensions	`[x for x in y]`	Complex logic

Usage Examples¶

Basic Evaluation¶

from flowengine import ConditionEvaluator, FlowContext

evaluator = ConditionEvaluator()
context = FlowContext()
context.set("status", "active")
context.set("count", 42)

# Evaluate conditions
result = evaluator.evaluate(
    "context.data.status == 'active'",
    context
)
print(result)  # True

result = evaluator.evaluate(
    "context.data.count > 100",
    context
)
print(result)  # False

Safety Validation¶

evaluator = ConditionEvaluator()

# Check if safe
print(evaluator.is_safe("context.data.x == 1"))  # True
print(evaluator.is_safe("len(context.data.x)"))  # False

# Get validation errors
errors = evaluator.validate("import os")
print(errors)  # ["Import statements are not allowed"]

AST Validation¶

import ast
from flowengine import SafeASTValidator

# Parse expression
tree = ast.parse("x == 1 and y > 2", mode="eval")

# Validate
validator = SafeASTValidator()
is_safe = validator.validate(tree)
errors = validator.get_errors()

if not is_safe:
    print("Validation errors:", errors)

Custom Evaluator with Engine¶

from flowengine import FlowEngine, ConditionEvaluator

# Create custom evaluator
evaluator = ConditionEvaluator()

# Use with engine
engine = FlowEngine(config, components, evaluator=evaluator)