wehub-resource-sync
fed8b2eed7
Backend release / release (push) Waiting to run
Bandit Security Scan / bandit_scan (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / build (linux/amd64, ubuntu-latest, amd64) (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / manifest (push) Blocked by required conditions
Build and push DocsGPT FE Docker image for development / build (linux/amd64, ubuntu-latest, amd64) (push) Waiting to run
Build and push DocsGPT FE Docker image for development / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Waiting to run
Build and push DocsGPT FE Docker image for development / manifest (push) Blocked by required conditions
Python linting / ruff (push) Waiting to run
Run python tests with pytest / Run tests and count coverage (3.12) (push) Waiting to run
React Widget Build / build (push) Waiting to run
65 lines
2.1 KiB
Python
65 lines
2.1 KiB
Python
from typing import Any, Dict
|
|
|
|
import celpy
|
|
import celpy.celtypes
|
|
|
|
|
|
class CelEvaluationError(Exception):
|
|
pass
|
|
|
|
|
|
def _convert_value(value: Any) -> Any:
|
|
if isinstance(value, bool):
|
|
return celpy.celtypes.BoolType(value)
|
|
if isinstance(value, int):
|
|
return celpy.celtypes.IntType(value)
|
|
if isinstance(value, float):
|
|
return celpy.celtypes.DoubleType(value)
|
|
if isinstance(value, str):
|
|
return celpy.celtypes.StringType(value)
|
|
if isinstance(value, list):
|
|
return celpy.celtypes.ListType([_convert_value(item) for item in value])
|
|
if isinstance(value, dict):
|
|
return celpy.celtypes.MapType(
|
|
{celpy.celtypes.StringType(k): _convert_value(v) for k, v in value.items()}
|
|
)
|
|
if value is None:
|
|
return celpy.celtypes.BoolType(False)
|
|
return celpy.celtypes.StringType(str(value))
|
|
|
|
|
|
def build_activation(state: Dict[str, Any]) -> Dict[str, Any]:
|
|
return {k: _convert_value(v) for k, v in state.items()}
|
|
|
|
|
|
def evaluate_cel(expression: str, state: Dict[str, Any]) -> Any:
|
|
if not expression or not expression.strip():
|
|
raise CelEvaluationError("Empty expression")
|
|
try:
|
|
env = celpy.Environment()
|
|
ast = env.compile(expression)
|
|
program = env.program(ast)
|
|
activation = build_activation(state)
|
|
result = program.evaluate(activation)
|
|
except celpy.CELEvalError as exc:
|
|
raise CelEvaluationError(f"CEL evaluation error: {exc}") from exc
|
|
except Exception as exc:
|
|
raise CelEvaluationError(f"CEL error: {exc}") from exc
|
|
return cel_to_python(result)
|
|
|
|
|
|
def cel_to_python(value: Any) -> Any:
|
|
if isinstance(value, celpy.celtypes.BoolType):
|
|
return bool(value)
|
|
if isinstance(value, celpy.celtypes.IntType):
|
|
return int(value)
|
|
if isinstance(value, celpy.celtypes.DoubleType):
|
|
return float(value)
|
|
if isinstance(value, celpy.celtypes.StringType):
|
|
return str(value)
|
|
if isinstance(value, celpy.celtypes.ListType):
|
|
return [cel_to_python(item) for item in value]
|
|
if isinstance(value, celpy.celtypes.MapType):
|
|
return {str(k): cel_to_python(v) for k, v in value.items()}
|
|
return value
|