Files
state-hub/task_flow_engine/builtins.py

38 lines
1021 B
Python

from __future__ import annotations
from collections.abc import Sequence
from typing import Any
EMPTY_VALUES = (None, "", [], {}, ())
def all_eq(values: list[Any], expected: Any) -> bool:
return all(_matches(value, expected) for value in values)
def any_eq(values: list[Any], expected: Any) -> bool:
return any(_matches(value, expected) for value in values)
def none_eq(values: list[Any], expected: Any) -> bool:
return all(not _matches(value, expected) for value in values)
def exists(values: list[Any], expected: Any = None) -> bool:
return any(value not in EMPTY_VALUES for value in values)
def count_gte(values: list[Any], expected: Any) -> bool:
try:
threshold = int(expected)
except (TypeError, ValueError):
return False
return len(values) >= threshold
def _matches(value: Any, expected: Any) -> bool:
if isinstance(expected, Sequence) and not isinstance(expected, (str, bytes, bytearray)):
return value in expected
return value == expected