Coverage for src/invariant/expressions.py: 93.71%

1"""CEL expression engine for evaluating ${...} expressions in node parameters.

3This module provides expression evaluation using the Common Expression Language (CEL).

4Expressions are embedded in parameter values using ${...} delimiters and are evaluated

5with dependency artifacts exposed as variables.

7Expression Syntax

8-----------------

10Variable References:

11 - `${x}` - References the entire artifact. For artifacts with a `.value` attribute,

12 this returns the value directly.

13 - `${x.value}` - Accesses the `value` field of artifact `x`. This is the most common

14 pattern for accessing numeric or string values from artifacts.

16Integer Arithmetic:

17 - `${x.value + 1}` - Add integers

18 - `${x.value + y.value}` - Add two artifact values

19 - `${x.value * 2}` - Multiply

20 - `${x.value - y.value}` - Subtract

22Decimal Arithmetic (Avoiding Floats):

23 Per the Strict Numeric Policy, native float types are forbidden in cacheable data

24 because IEEE 754 floats are non-deterministic across architectures. Use the `decimal()`

25 function for fractional arithmetic:

27 - `${decimal("3.14")}` - Create a Decimal from a string

28 - `${decimal("1.5") + decimal("2.5")}` - Decimal arithmetic (returns Decimal)

29 - `${decimal(x.value)}` - Convert an integer to Decimal

30 - `${decimal("3.14") * 2}` - Decimal multiplication

32Built-in Functions:

33 - `min(a, b)` - Returns the minimum of two comparable values

34 - `max(a, b)` - Returns the maximum of two comparable values

35 - `decimal(value)` - Converts a value (int, string, or Decimal) to Decimal

37 Examples:

38 - `${min(x, y)}` - Returns the artifact with the smaller value

39 - `${min(x.value, y.value)}` - Returns the smaller numeric value

40 - `${max(x.value, 10)}` - Returns the larger of x.value or 10

42String Interpolation:

43 - `"prefix_${x.value}_suffix"` - Mixed text and expressions

44 - `"Result: ${x.value + y.value}"` - String concatenation with expressions

46Error Cases:

47 - Expressions that return float (double) values raise ValueError. Use `decimal()`

48 for fractional arithmetic.

49 - References to undeclared dependencies raise ValueError.

50 - Invalid expression syntax raises ValueError with details.

52Examples

53--------

55Basic variable access:

56 >>> params = {"width": "${background}"}

57 >>> deps = {"background": 100}

58 >>> resolve_params(params, deps)

59 {"width": 100}

61Arithmetic with multiple dependencies:

62 >>> params = {"sum": "${x + y}"}

63 >>> deps = {"x": 3, "y": 7}

64 >>> resolve_params(params, deps)

65 {"sum": 10}

67Decimal arithmetic (avoiding floats):

68 >>> params = {"result": "${decimal(\"3.14\") + decimal(\"2.86\")}"}

69 >>> deps = {}

70 >>> resolve_params(params, deps)

71 {"result": Decimal("6.00")}

73Using min/max for canonicalization:

74 >>> params = {"a": "${min(x, y)}", "b": "${max(x, y)}"}

75 >>> deps = {"x": 7, "y": 3}

76 >>> resolved = resolve_params(params, deps)

77 >>> resolved["a"] # Returns 3 (the smaller value)

78 3

79 >>> resolved["b"] # Returns 7 (the larger value)

80 7

82String interpolation:

83 >>> params = {"message": "Width is ${width}px"}

84 >>> deps = {"width": 200}

85 >>> resolve_params(params, deps)

86 {"message": "Width is 200px"}

87"""

89import re

90from decimal import Decimal

91from functools import lru_cache

92from typing import Any

94import celpy

95import celpy.celparser

96import celpy.celtypes as celtypes

97import celpy.evaluation

99from invariant.params import cel, ref

100from invariant.protocol import ICacheable

101

102

103def resolve_params(

104 params: dict[str, Any], dependencies: dict[str, Any]

105) -> dict[str, Any]:

106 """Resolve ${...} CEL expressions in parameter values.

107

108 Walks through the params dictionary, finds values containing ${...} delimiters,

109 evaluates the CEL expression with dependency artifacts exposed as variables,

110 and replaces the expression with the resolved value.

111

112 Args:

113 params: Dictionary of parameter name -> value. Values may contain ${...} expressions.

114 dependencies: Dictionary mapping dependency IDs to their artifacts (native types or ICacheable).

115 These are exposed as variables in CEL expressions.

116

117 Returns:

118 Dictionary with all ${...} expressions resolved to their evaluated values.

119

120 Raises:

121 ValueError: If expression evaluation fails, undeclared dependencies are referenced,

122 or expression result is a float (double).

123 """

124 resolved = {}

125 for key, value in params.items():

126 resolved[key] = _resolve_value(value, dependencies)

127 return resolved

128

129

130def _resolve_value(value: Any, dependencies: dict[str, Any]) -> Any:

131 """Recursively resolve expressions in a value.

132

133 Handles ref() markers (artifact passthrough), cel() markers (CEL expressions),

134 ${...} string interpolation, and nested structures (dicts, lists).

135

136 Args:

137 value: The value to resolve. May be:

138 - ref(dep): Resolves to the artifact from dependency (native type or ICacheable)

139 - cel(expr): Evaluates CEL expression and returns computed value

140 - str with ${...}: String interpolation with CEL expressions

141 - dict/list: Recursively resolves nested values

142 - Other: Returns as-is

143 dependencies: Dictionary of dependency artifacts (native types or ICacheable).

144

145 Returns:

146 Resolved value with all markers and expressions evaluated.

147

148 Raises:

149 ValueError: If ref() references undeclared dependency.

150 """

151 # Handle ref() marker - artifact passthrough

152 if isinstance(value, ref):

153 if value.dep not in dependencies:

154 raise ValueError(

155 f"ref('{value.dep}') references undeclared dependency '{value.dep}'. "

156 f"Available dependencies: {list(dependencies.keys())}"

157 )

158 return dependencies[value.dep]

159

160 # Handle cel() marker - CEL expression evaluation

161 if isinstance(value, cel):

162 return _evaluate_cel(value.expr, dependencies)

163

164 # Handle string with ${...} interpolation

165 if isinstance(value, str):

166 if "${" in value and "}" in value:

167 # Extract and evaluate CEL expression

168 return _evaluate_expression(value, dependencies)

169 return value

170

171 # Handle nested structures

172 if isinstance(value, dict):

173 return {k: _resolve_value(v, dependencies) for k, v in value.items()}

174 if isinstance(value, list):

175 return [_resolve_value(item, dependencies) for item in value]

176

177 # Primitive value, return as-is

178 return value

179

180

181def _evaluate_expression(expr_string: str, dependencies: dict[str, Any]) -> Any:

182 """Evaluate a CEL expression string.

183

184 Args:

185 expr_string: String potentially containing ${...} expressions.

186 dependencies: Dictionary of dependency artifacts.

187

188 Returns:

189 Resolved value after evaluating all expressions.

190

191 Raises:

192 ValueError: If expression is invalid or references undeclared dependencies.

193 """

194 # Find all ${...} expressions

195 pattern = r"\$\{([^}]+)\}"

196 matches = re.findall(pattern, expr_string)

197

198 if not matches:

199 # No expressions found, return as-is

200 return expr_string

201

202 if len(matches) == 1 and expr_string.strip() == f"${{{matches[0]}}}":

203 # Single expression covering the entire string - evaluate it

204 cel_expr = matches[0].strip()

205 return _evaluate_cel(cel_expr, dependencies)

206 else:

207 # Multiple expressions or mixed with text - evaluate each and substitute

208 result = expr_string

209 for match in matches:

210 cel_expr = match.strip()

211 evaluated = _evaluate_cel(cel_expr, dependencies)

212 # Convert to string for substitution

213 result = result.replace(f"${{{match}}}", str(evaluated))

214 return result

215

216

217@lru_cache(maxsize=1024)

218def _compiled_expression(expression):

219 # Cache syntax only; programs and dependency activations remain per call.

220 return celpy.Environment().compile(expression)

221

222

223def _evaluate_cel(expression: str, dependencies: dict[str, Any]) -> Any:

224 """Evaluate a single CEL expression.

225

226 Args:

227 expression: The CEL expression to evaluate (without ${...} delimiters).

228 dependencies: Dictionary of dependency artifacts (native types or ICacheable).

229

230 Returns:

231 The evaluated result. Must be a type that can be hashed (no floats).

232

233 Raises:

234 ValueError: If expression is invalid, references undeclared deps, or returns float.

235 """

236 # Build CEL environment

237 env = celpy.Environment()

238

239 # Convert dependencies to CEL-compatible types

240 var_declarations: dict[str, Any] = {}

241 for dep_id, artifact in dependencies.items():

242 # Convert artifact to CEL-compatible type

243 cel_value = _value_to_cel(artifact)

244 var_declarations[dep_id] = cel_value

245

246 # Compile the expression

247 try:

248 ast = (

249 _compiled_expression(expression)

250 if len(expression) <= 4096

251 else env.compile(expression)

252 )

253 except celpy.celparser.CELParseError as e:

254 raise ValueError(f"Failed to parse CEL expression '{expression}': {e}") from e

255

256 # Register custom functions: decimal(), min(), max()

257 custom_functions = {

258 "decimal": _decimal_function,

259 "min": _min_function,

260 "max": _max_function,

261 }

262

263 # Create program with custom functions

264 program = env.program(ast, functions=custom_functions)

265

266 # Create activation with variables

267 activation: dict[str, Any] = {}

268 for var_name, var_value in var_declarations.items():

269 activation[var_name] = var_value

270

271 # Evaluate

272 try:

273 result = program.evaluate(activation)

274 except celpy.evaluation.CELEvalError as e:

275 raise ValueError(

276 f"Failed to evaluate CEL expression '{expression}': {e}"

277 ) from e

278

279 # Check for float results (forbidden per Strict Numeric Policy)

280 if isinstance(result, float):

281 raise ValueError(

282 f"Expression '{expression}' returned a float (double), which is forbidden. "

283 f"Use decimal() for fractional arithmetic."

284 )

285

286 # If result is a MapType (from ICacheable domain type), extract the value for simple variable references

287 if isinstance(result, celtypes.MapType):

288 # Check if this is a simple variable reference like ${x}

289 # If the map has a 'value' key, extract it

290 value_key = celtypes.StringType("value")

291 if value_key in result:

292 result = result[value_key]

293 else:

294 # Return the map as-is (might be used in further expressions)

295 return _cel_to_python(result, expression)

296

297 # Convert result to appropriate Python type

298 return _cel_to_python(result, expression)

299

300

301def _value_to_cel(value: Any) -> Any:

302 """Convert a value (native type or ICacheable) to a CEL-compatible type.

303

304 Native types are exposed directly. ICacheable domain types are converted to MapType

305 for field access.

306

307 Args:

308 value: The value to convert (int, str, Decimal, dict, list, ICacheable, etc.).

309

310 Returns:

311 CEL-compatible value (IntType, StringType, MapType, etc.).

312 """

313 # Native types - expose directly

314 if isinstance(value, int):

315 return celtypes.IntType(value)

316 if isinstance(value, str):

317 return celtypes.StringType(value)

318 if isinstance(value, bool):

319 return celtypes.BoolType(value)

320 if isinstance(value, Decimal):

321 # Decimal is passed through as-is (CEL doesn't have native Decimal)

322 return value

323 if isinstance(value, dict):

324 # Convert dict to CEL MapType recursively

325 result: dict[celtypes.StringType, Any] = {}

326 for key, val in value.items():

327 result[celtypes.StringType(key)] = _value_to_cel(val)

328 return celtypes.MapType(result)

329 if isinstance(value, (list, tuple)):

330 # Convert list/tuple to CEL ListType recursively

331 return celtypes.ListType([_value_to_cel(item) for item in value])

332 if value is None:

333 # None is represented as null in CEL

334 return None

335

336 # ICacheable domain types - convert to MapType for field access

337 if isinstance(value, ICacheable):

338 return _icacheable_to_cel_map(value)

339

340 # Fallback: convert to string

341 return celtypes.StringType(str(value))

342

343

344def _icacheable_to_cel_map(artifact: ICacheable) -> celtypes.MapType:

345 """Convert an ICacheable domain type to a CEL-compatible MapType.

346

347 Exposes all public attributes of the artifact for field access.

348

349 Args:

350 artifact: The ICacheable artifact to convert.

351

352 Returns:

353 MapType representation suitable for CEL evaluation.

354 """

355 result: dict[celtypes.StringType, Any] = {}

356

357 # Expose all public attributes

358 for attr_name in dir(artifact):

359 if not attr_name.startswith("_") and not callable(

360 getattr(artifact, attr_name, None)

361 ):

362 try:

363 attr_value = getattr(artifact, attr_name)

364 # Convert attribute value to CEL type recursively

365 result[celtypes.StringType(attr_name)] = _value_to_cel(attr_value)

366 except AttributeError:

367 pass

368

369 return celtypes.MapType(result)

370

371

372def _cel_to_python(cel_value: Any, expression: str) -> Any:

373 """Convert a CEL value to a Python type.

374

375 Handles conversion of CEL types (IntType, StringType, etc.) to Python types.

376

377 Args:

378 cel_value: The value returned from CEL evaluation.

379 expression: The original expression (for error messages).

380

381 Returns:

382 Python value (int, str, Decimal, ICacheable, etc.)

383

384 Raises:

385 ValueError: If value is a float (forbidden).

386 """

387 # Handle CEL types

388 if isinstance(cel_value, celtypes.IntType):

389 return int(cel_value)

390 elif isinstance(cel_value, celtypes.StringType):

391 return str(cel_value)

392 elif isinstance(cel_value, celtypes.BoolType):

393 return bool(cel_value)

394 elif isinstance(cel_value, float):

395 # This should have been caught earlier, but double-check

396 raise ValueError(

397 f"Expression '{expression}' returned float, which is forbidden. "

398 f"Use decimal() for fractional values."

399 )

400 elif isinstance(cel_value, Decimal):

401 return cel_value

402 elif isinstance(cel_value, celtypes.MapType):

403 # Convert MapType to dict

404 result_dict: dict[str, Any] = {}

405 for key, value in cel_value.items():

406 if isinstance(key, celtypes.StringType):

407 result_dict[str(key)] = _cel_to_python(value, expression)

408 else:

409 result_dict[str(key)] = _cel_to_python(value, expression)

410 return result_dict

411 elif isinstance(cel_value, (int, str, bool)):

412 # Already Python types

413 return cel_value

414 elif isinstance(cel_value, dict):

415 # Plain dict - convert recursively

416 return {k: _cel_to_python(v, expression) for k, v in cel_value.items()}

417 elif isinstance(cel_value, list):

418 # Plain list - convert recursively

419 return [_cel_to_python(item, expression) for item in cel_value]

420 else:

421 # Unknown type - return as-is (might be a CEL-specific type or already Python)

422 return cel_value

423

424

425# Custom CEL functions

426def _decimal_function(value: Any) -> Decimal:

427 """CEL function: decimal(value) - constructs a Decimal from int, string, or Decimal.

428

429 Args:

430 value: Can be IntType, StringType, int, str, Decimal, or MapType with value field.

431

432 Returns:

433 Decimal value.

434

435 Raises:

436 ValueError: If value cannot be converted to Decimal.

437 """

438 # Handle CEL types

439 if isinstance(value, celtypes.IntType):

440 return Decimal(str(int(value)))

441 elif isinstance(value, celtypes.StringType):

442 return Decimal(str(value))

443 elif isinstance(value, celtypes.MapType):

444 # Extract value from MapType

445 value_key = celtypes.StringType("value")

446 if value_key in value:

447 return _decimal_function(value[value_key])

448 else:

449 raise ValueError("Cannot extract value from MapType for decimal conversion")

450 # Handle Python types

451 elif isinstance(value, Decimal):

452 return value

453 elif isinstance(value, int):

454 return Decimal(str(value))

455 elif isinstance(value, str):

456 return Decimal(value)

457 elif isinstance(value, ICacheable) and hasattr(value, "value"):

458 # Extract value from ICacheable

459 return Decimal(str(value.value))

460 else:

461 raise ValueError(f"Cannot convert {type(value)} to Decimal")

462

463

464def _min_function(a: Any, b: Any) -> Any:

465 """CEL function: min(a, b) - returns the minimum of two comparable values.

466

467 Handles both direct values and MapType artifacts (extracts .value for comparison).

468

469 Args:

470 a: First value (can be IntType, StringType, MapType, etc.)

471 b: Second value (can be IntType, StringType, MapType, etc.)

472

473 Returns:

474 The minimum of a and b. If both are MapType, extracts .value for comparison.

475 """

476 # Handle MapType arguments - extract value for comparison

477 a_val = _extract_comparison_value(a)

478 b_val = _extract_comparison_value(b)

479

480 # Compare and return the original type (not the extracted value)

481 if a_val < b_val:

482 return a

483 return b

484

485

486def _max_function(a: Any, b: Any) -> Any:

487 """CEL function: max(a, b) - returns the maximum of two comparable values.

488

489 Handles both direct values and MapType artifacts (extracts .value for comparison).

490

491 Args:

492 a: First value (can be IntType, StringType, MapType, etc.)

493 b: Second value (can be IntType, StringType, MapType, etc.)

494

495 Returns:

496 The maximum of a and b. If both are MapType, extracts .value for comparison.

497 """

498 # Handle MapType arguments - extract value for comparison

499 a_val = _extract_comparison_value(a)

500 b_val = _extract_comparison_value(b)

501

502 # Compare and return the original type (not the extracted value)

503 if a_val > b_val:

504 return a

505 return b

506

507

508def _extract_comparison_value(value: Any) -> Any:

509 """Extract a comparable value from a CEL type for min/max comparison.

510

511 Args:

512 value: Can be MapType, IntType, StringType, or other comparable type.

513

514 Returns:

515 A value suitable for comparison (<, >, etc.)

516 """

517 if isinstance(value, celtypes.MapType):

518 # Extract .value field from MapType

519 value_key = celtypes.StringType("value")

520 if value_key in value:

521 return _extract_comparison_value(value[value_key])

522 else:

523 # If no value field, try to compare the map itself (not ideal but works)

524 return value

525 elif isinstance(value, celtypes.IntType):

526 return int(value)

527 elif isinstance(value, celtypes.StringType):

528 return str(value)

529 elif isinstance(value, celtypes.BoolType):

530 return bool(value)

531 else:

532 # Already a comparable Python type

533 return value

Coverage for src / invariant / expressions.py: 93.71%

175 statements