"""Closed, typed rule expressions for governed DataOps rules. The module deliberately owns both parsing and validation. It produces JSON only; expressions are never delegated to Python, SQL, JavaScript, or a model runtime. Compilers consume the resulting AST in a later release. """ from __future__ import annotations from dataclasses import dataclass from datetime import date as Date from datetime import datetime from decimal import Decimal, InvalidOperation, ROUND_HALF_UP import re from typing import Any from zoneinfo import ZoneInfo MAX_SOURCE_LENGTH = 20_000 MAX_AST_DEPTH = 40 MAX_AST_NODES = 500 MAX_REGEX_LENGTH = 500 ALLOWED_FUNCTIONS = frozenset( { "matches", "lower", "upper", "trim", "length", "coalesce", "date", "timestamp", "abs", "round", } ) SUPPORTED_BACKENDS = frozenset({"postgresql", "mysql", "polars"}) _FUNCTION_BACKENDS = { "matches": SUPPORTED_BACKENDS, # The reference oracle verifies Python/Polars Unicode case folding. SQL # engines require a deployment-pinned collation before they may opt in. "lower": frozenset({"polars"}), "upper": frozenset({"polars"}), "trim": SUPPORTED_BACKENDS, "length": SUPPORTED_BACKENDS, "coalesce": SUPPORTED_BACKENDS, "date": SUPPORTED_BACKENDS, # MySQL timezone conversion depends on server timezone tables; do not # advertise it before a deployment can prove that dependency. "timestamp": frozenset({"postgresql", "polars"}), "abs": SUPPORTED_BACKENDS, # Polars rounding mode cannot be assumed to match the V1 decimal policy. "round": frozenset({"postgresql", "mysql"}), } _IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,127}$") _FIELD_TYPE_ALIASES = { "bigint": "integer", "bool": "boolean", "character varying": "string", "datetime": "timestamp", "int": "integer", "numeric": "decimal", "text": "string", "varchar": "string", } _FIELD_TYPES = { "binary", "boolean", "date", "decimal", "double", "float", "integer", "json", "string", "timestamp", "timestamptz", } _NUMERIC_TYPES = {"integer", "decimal", "float", "double"} _LITERAL_TYPES = {"boolean", "decimal", "integer", "null", "string"} _DECIMAL_LEXEME = re.compile(r"^[0-9]+\.[0-9]+$") _BINARY_OPERATORS = { "||", "&&", "==", "!=", "<", "<=", ">", ">=", "+", "-", "*", "/", "%", } _PRECEDENCE = { "||": 10, "&&": 20, "==": 30, "!=": 30, "<": 30, "<=": 30, ">": 30, ">=": 30, "+": 40, "-": 40, "*": 50, "/": 50, "%": 50, } @dataclass(frozen=True) class _Token: kind: str value: Any position: int def _error(position: int, message: str) -> ValueError: return ValueError(f"expression {message} at position {position}") class _Lexer: def __init__(self, source: str): self.source = source self.length = len(source) self.position = 0 def tokens(self) -> list[_Token]: result: list[_Token] = [] while self.position < self.length: char = self.source[self.position] if char.isspace(): self.position += 1 elif char in "'\"": result.append(self._string()) elif char.isdigit(): result.append(self._number()) elif char.isascii() and (char.isalpha() or char == "_"): result.append(self._identifier()) elif char in "(),": result.append(_Token(char, char, self.position)) self.position += 1 else: result.append(self._operator()) result.append(_Token("EOF", None, self.position)) return result def _string(self) -> _Token: start = self.position quote = self.source[self.position] self.position += 1 pieces: list[str] = [] escapes = {"n": "\n", "r": "\r", "t": "\t", "b": "\b", "f": "\f"} while self.position < self.length: char = self.source[self.position] self.position += 1 if char == quote: value = "".join(pieces) if len(value) > 4_000: raise _error(start, "string literal exceeds 4000 characters") return _Token("LITERAL", ("string", value), start) if char != "\\": pieces.append(char) continue if self.position >= self.length: raise _error(start, "contains an unfinished string escape") escaped = self.source[self.position] self.position += 1 if escaped in escapes: pieces.append(escapes[escaped]) elif escaped in {"\\", "'", '\"'}: pieces.append(escaped) elif escaped == "u": encoded = self.source[self.position : self.position + 4] if len(encoded) != 4 or not re.fullmatch(r"[0-9a-fA-F]{4}", encoded): raise _error(start, "contains an invalid unicode escape") pieces.append(chr(int(encoded, 16))) self.position += 4 else: raise _error(start, "contains an unsupported string escape") raise _error(start, "contains an unterminated string") def _number(self) -> _Token: start = self.position while self.position < self.length and self.source[self.position].isdigit(): self.position += 1 kind = "integer" if self.position < self.length and self.source[self.position] == ".": kind = "decimal" self.position += 1 decimal_start = self.position while self.position < self.length and self.source[self.position].isdigit(): self.position += 1 if self.position == decimal_start: raise _error(start, "contains an invalid numeric literal") raw = self.source[start : self.position] # Decimal tokens keep their lexeme in the JSON AST. A float here # would destroy source precision before a compiler receives it. value: int | str = int(raw) if kind == "integer" else raw return _Token("LITERAL", (kind, value), start) def _identifier(self) -> _Token: start = self.position self.position += 1 while self.position < self.length: char = self.source[self.position] if not (char.isascii() and (char.isalnum() or char == "_")): break self.position += 1 value = self.source[start : self.position] if len(value) > 128: raise _error(start, "identifier exceeds 128 characters") lowered = value.lower() if lowered == "true": return _Token("LITERAL", ("boolean", True), start) if lowered == "false": return _Token("LITERAL", ("boolean", False), start) if lowered == "null": return _Token("LITERAL", ("null", None), start) return _Token("IDENT", value, start) def _operator(self) -> _Token: start = self.position for operator in ("&&", "||", "==", "!=", "<=", ">="): if self.source.startswith(operator, start): self.position += len(operator) return _Token("OP", operator, start) char = self.source[self.position] if char in "!<>+-*/%": self.position += 1 return _Token("OP", char, start) raise _error(start, f"contains unexpected character {char!r}") class _Parser: def __init__(self, source: str): self.tokens = _Lexer(source).tokens() self.index = 0 @property def current(self) -> _Token: return self.tokens[self.index] def consume(self, kind: str, value: str | None = None) -> _Token: token = self.current if token.kind != kind or (value is not None and token.value != value): expected = value if value is not None else kind raise _error(token.position, f"expected {expected!r}") self.index += 1 return token def parse(self) -> dict[str, Any]: expression = self._expression(0) if self.current.kind != "EOF": raise _error(self.current.position, "contains unexpected trailing input") return expression def _expression(self, minimum_precedence: int) -> dict[str, Any]: token = self.current self.index += 1 if token.kind == "LITERAL": literal_type, value = token.value left = {"kind": "literal", "type": literal_type, "value": value} elif token.kind == "IDENT": if self.current.kind == "(": if token.value not in ALLOWED_FUNCTIONS: raise _error( token.position, f"uses unsupported expression function {token.value!r}", ) self.index += 1 arguments: list[dict[str, Any]] = [] if self.current.kind != ")": while True: arguments.append(self._expression(0)) if self.current.kind != ",": break self.index += 1 self.consume(")") left = { "kind": "call", "function": token.value, "arguments": arguments, } else: left = {"kind": "identifier", "name": token.value} elif token.kind == "OP" and token.value in {"!", "-"}: left = { "kind": "unary", "operator": token.value, "operand": self._expression(60), } elif token.kind == "(": left = self._expression(0) self.consume(")") else: raise _error(token.position, "contains an unexpected token") while self.current.kind == "OP": operator = self.current.value precedence = _PRECEDENCE.get(operator) if precedence is None or precedence < minimum_precedence: break self.index += 1 left = { "kind": "binary", "operator": operator, "left": left, "right": self._expression(precedence + 1), } return left def parse_expression(source: str) -> dict: """Parse a V1 expression into the closed, JSON-serializable AST.""" if not isinstance(source, str) or not source.strip(): raise ValueError("expression source is required") if len(source) > MAX_SOURCE_LENGTH: raise ValueError(f"expression source exceeds {MAX_SOURCE_LENGTH} characters") ast = _Parser(source).parse() _validate_ast(ast) return ast def _validate_ast(ast: Any) -> None: count = 0 def walk(node: Any, depth: int) -> None: nonlocal count if depth > MAX_AST_DEPTH: raise ValueError(f"expression AST depth exceeds {MAX_AST_DEPTH}") count += 1 if count > MAX_AST_NODES: raise ValueError(f"expression AST node count exceeds {MAX_AST_NODES}") if not isinstance(node, dict): raise ValueError("expression AST node must be an object") kind = node.get("kind") if kind == "literal": if set(node) != {"kind", "type", "value"}: raise ValueError("literal AST node must have a closed shape") literal_type = node.get("type") value = node.get("value") if literal_type not in _LITERAL_TYPES: raise ValueError("literal AST node has an unsupported type") if literal_type == "null" and value is not None: raise ValueError("null literal must have a null value") if literal_type == "boolean" and not isinstance(value, bool): raise ValueError("boolean literal must have a boolean value") if literal_type == "string" and ( not isinstance(value, str) or len(value) > 4_000 ): raise ValueError("string literal is invalid") if literal_type == "integer" and ( isinstance(value, bool) or not isinstance(value, int) ): raise ValueError("integer literal is invalid") if literal_type == "decimal" and ( not isinstance(value, str) or _DECIMAL_LEXEME.fullmatch(value) is None ): raise ValueError("decimal literal is invalid") return if kind == "identifier": if set(node) != {"kind", "name"} or not isinstance(node.get("name"), str): raise ValueError("identifier AST node must have a closed shape") if _IDENTIFIER.fullmatch(node["name"]) is None: raise ValueError("identifier AST node has an invalid name") return if kind == "unary": if set(node) != {"kind", "operator", "operand"}: raise ValueError("unary AST node must have a closed shape") if node.get("operator") not in {"!", "-"}: raise ValueError("unary AST node has an unsupported operator") walk(node.get("operand"), depth + 1) return if kind == "binary": if set(node) != {"kind", "operator", "left", "right"}: raise ValueError("binary AST node must have a closed shape") if node.get("operator") not in _BINARY_OPERATORS: raise ValueError("binary AST node has an unsupported operator") walk(node.get("left"), depth + 1) walk(node.get("right"), depth + 1) return if kind == "call": if set(node) != {"kind", "function", "arguments"}: raise ValueError("call AST node must have a closed shape") function = node.get("function") arguments = node.get("arguments") if not isinstance(function, str) or _IDENTIFIER.fullmatch(function) is None: raise ValueError("call AST node has an invalid function") if function not in ALLOWED_FUNCTIONS: raise ValueError(f"unsupported expression function: {function}") if not isinstance(arguments, list) or len(arguments) > 100: raise ValueError("call AST arguments must be a bounded array") for argument in arguments: walk(argument, depth + 1) # This resource bound is independent of schema typing, so RuleSpec # normalization can reject an unsafe regex before compilation. if ( function == "matches" and len(arguments) >= 2 and arguments[1].get("kind") == "literal" and arguments[1].get("type") == "string" ): if len(arguments[1]["value"]) > MAX_REGEX_LENGTH: raise ValueError( f"regex pattern exceeds {MAX_REGEX_LENGTH} characters" ) return raise ValueError("expression AST node kind is unsupported") walk(ast, 1) def _normalized_fields(fields: Any) -> dict[str, str]: if not isinstance(fields, dict) or len(fields) > 1_000: raise ValueError("expression fields must be a bounded object") normalized: dict[str, str] = {} for name, field_type in fields.items(): if not isinstance(name, str) or _IDENTIFIER.fullmatch(name) is None: raise ValueError("expression field names must be identifiers") if not isinstance(field_type, str): raise ValueError(f"expression field {name} type is invalid") result_type = _FIELD_TYPE_ALIASES.get( field_type.strip().lower(), field_type.strip().lower() ) if result_type not in _FIELD_TYPES: raise ValueError(f"expression field {name} has an unsupported type") normalized[name] = result_type return normalized def _is_numeric(value_type: str) -> bool: return value_type in _NUMERIC_TYPES def _promote_numeric(left: str, right: str) -> str: order = {"integer": 0, "decimal": 1, "float": 2, "double": 3} return max((left, right), key=lambda value: order[value]) def _compatible(left: str, right: str) -> bool: return left == "null" or right == "null" or left == right or ( _is_numeric(left) and _is_numeric(right) ) def _require_arguments(function: str, arguments: list[dict], expected: int | tuple[int, int]) -> None: if isinstance(expected, int): valid = len(arguments) == expected else: valid = expected[0] <= len(arguments) <= expected[1] if not valid: raise ValueError(f"function {function} received an unsupported argument count") def type_check_expression(ast: dict, fields: dict[str, str]) -> str: """Validate schema-bound identifiers and return the expression result type.""" _validate_ast(ast) normalized_fields = _normalized_fields(fields) def check(node: dict) -> str: kind = node["kind"] if kind == "literal": return node["type"] if kind == "identifier": name = node["name"] if name not in normalized_fields: raise ValueError(f"unknown expression field: {name}") return normalized_fields[name] if kind == "unary": operand_type = check(node["operand"]) operator = node["operator"] if operator == "!": if operand_type != "boolean": raise ValueError("logical not requires a boolean operand") return "boolean" if not _is_numeric(operand_type): raise ValueError("numeric negation requires a numeric operand") return operand_type if kind == "binary": left_type = check(node["left"]) right_type = check(node["right"]) operator = node["operator"] if operator in {"&&", "||"}: if left_type != "boolean" or right_type != "boolean": raise ValueError("logical operators require boolean operands") return "boolean" if operator in {"+", "-", "*", "/", "%"}: if not _is_numeric(left_type) or not _is_numeric(right_type): raise ValueError("arithmetic operators require numeric operands") return _promote_numeric(left_type, right_type) if not _compatible(left_type, right_type): raise ValueError( "comparison operands must have compatible types; use an explicit cast" ) return "boolean" function = node["function"] arguments = node["arguments"] if function not in ALLOWED_FUNCTIONS: raise ValueError(f"unsupported expression function: {function}") argument_types = [check(argument) for argument in arguments] if function == "matches": _require_arguments(function, arguments, 2) if argument_types != ["string", "string"]: raise ValueError("matches requires string arguments") pattern = arguments[1] if pattern["kind"] != "literal" or pattern["type"] != "string": raise ValueError("matches pattern must be a string literal") if len(pattern["value"]) > MAX_REGEX_LENGTH: raise ValueError(f"regex pattern exceeds {MAX_REGEX_LENGTH} characters") try: re.compile(pattern["value"]) except re.error as exc: raise ValueError("regex pattern is invalid") from exc return "boolean" if function in {"lower", "upper", "trim"}: _require_arguments(function, arguments, 1) if argument_types != ["string"]: raise ValueError(f"{function} requires a string argument") return "string" if function == "length": _require_arguments(function, arguments, 1) if argument_types != ["string"]: raise ValueError("length requires a string argument") return "integer" if function == "coalesce": _require_arguments(function, arguments, (2, 100)) concrete = [value for value in argument_types if value != "null"] if not concrete: return "null" if not all(_compatible(concrete[0], value) for value in concrete[1:]): raise ValueError("coalesce arguments must have compatible types") if all(_is_numeric(value) for value in concrete): return max(concrete, key=lambda value: {"integer": 0, "decimal": 1, "float": 2, "double": 3}[value]) return concrete[0] if function == "date": _require_arguments(function, arguments, 1) if argument_types[0] not in {"string", "date", "timestamp", "timestamptz"}: raise ValueError("date requires a string or temporal argument") return "date" if function == "timestamp": _require_arguments(function, arguments, 1) if argument_types[0] not in {"string", "date", "timestamp", "timestamptz"}: raise ValueError("timestamp requires a string or temporal argument") return "timestamp" if function == "abs": _require_arguments(function, arguments, 1) if not _is_numeric(argument_types[0]): raise ValueError("abs requires a numeric argument") return argument_types[0] _require_arguments(function, arguments, (1, 2)) if not _is_numeric(argument_types[0]): raise ValueError("round requires a numeric value") if len(argument_types) == 2 and argument_types[1] != "integer": raise ValueError("round precision requires an integer") return argument_types[0] return check(ast) def validate_rule_expressions( rule_spec: dict[str, Any], fields: dict[str, str] ) -> frozenset[str]: """Type-check every canonical expression in a schema-bound RuleSpec. Callers must supply fields from a server-owned schema snapshot, never an authoring request. The return value is immutable-plan metadata rather than executable code and lets release reject unsupported backends early. """ if not isinstance(rule_spec, dict) or not isinstance( rule_spec.get("steps"), list ): raise ValueError("rule spec steps are required for expression validation") supported_for_rule = SUPPORTED_BACKENDS for step in rule_spec["steps"]: if not isinstance(step, dict): raise ValueError("rule step must be an object") operation = step.get("op") ast = step.get("expression_ast") if operation in {"assert", "filter"} and ast is None: raise ValueError(f"{operation} expression is required") if ast is None: continue expression_type = type_check_expression(ast, fields) if operation in {"assert", "filter"} and expression_type != "boolean": raise ValueError(f"{operation} expression must return boolean") supported = backend_support(ast) if not supported: raise ValueError("expression has no supported execution backend") supported_for_rule = supported_for_rule & supported return frozenset(supported_for_rule) def _portable_regex(pattern: str) -> bool: """Keep only the portable regular-expression subset advertised by V1.""" if len(pattern) > MAX_REGEX_LENGTH: return False if re.search(r"\(\?|\\[1-9]|\\[pP]", pattern): return False try: re.compile(pattern) except re.error: return False return True def backend_support(ast: dict) -> frozenset[str]: """Return backends that can preserve this grammar's V1 semantics. The result is intentionally a capability analysis, not a compiler. It validates the closed AST again so hand-authored JSON cannot bypass the function allowlist before a compiler sees it. """ _validate_ast(ast) def support(node: dict) -> frozenset[str]: kind = node["kind"] if kind in {"literal", "identifier"}: return SUPPORTED_BACKENDS if kind == "unary": return support(node["operand"]) if kind == "binary": return support(node["left"]) & support(node["right"]) function = node["function"] supported = _FUNCTION_BACKENDS[function] for argument in node["arguments"]: supported = supported & support(argument) if function == "matches": arguments = node["arguments"] if ( len(arguments) != 2 or arguments[1]["kind"] != "literal" or arguments[1]["type"] != "string" or not _portable_regex(arguments[1]["value"]) ): return frozenset() return frozenset(supported) return support(ast) def _decimal(value: Any) -> Decimal: if isinstance(value, Decimal): return value if isinstance(value, bool): raise ValueError("reference evaluator expected a numeric value") try: if isinstance(value, float): return Decimal(str(value)) return Decimal(value) except (InvalidOperation, TypeError, ValueError) as exc: raise ValueError("reference evaluator expected a numeric value") from exc def _timestamp(value: Any, timezone: ZoneInfo) -> datetime: if isinstance(value, datetime): result = value elif isinstance(value, Date): result = datetime.combine(value, datetime.min.time()) elif isinstance(value, str): try: result = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as exc: raise ValueError("reference evaluator received an invalid timestamp") from exc else: raise ValueError("reference evaluator expected a temporal value") if result.tzinfo is None: result = result.replace(tzinfo=timezone) return result def _sql_and(left: Any, right: Any) -> bool | None: if left is False or right is False: return False if left is None or right is None: return None return bool(left and right) def _sql_or(left: Any, right: Any) -> bool | None: if left is True or right is True: return True if left is None or right is None: return None return bool(left or right) def evaluate_expression( ast: dict, row: dict[str, Any], *, timezone: str = "UTC" ) -> Any: """Execute only the closed AST as a deterministic semantic test oracle. This is intentionally not a production runner. It has no access to a filesystem, network, Python callables, SQL, or model-generated code. """ _validate_ast(ast) if not isinstance(row, dict): raise ValueError("reference evaluator row must be an object") try: timezone_info = ZoneInfo(timezone) except Exception as exc: raise ValueError("reference evaluator timezone is invalid") from exc def evaluate(node: dict) -> Any: kind = node["kind"] if kind == "literal": if node["type"] == "decimal": return Decimal(node["value"]) return node["value"] if kind == "identifier": if node["name"] not in row: raise ValueError(f"reference evaluator missing field: {node['name']}") return row[node["name"]] if kind == "unary": value = evaluate(node["operand"]) if value is None: return None return (not value) if node["operator"] == "!" else -_decimal(value) if kind == "binary": left = evaluate(node["left"]) right = evaluate(node["right"]) operator = node["operator"] if operator == "&&": return _sql_and(left, right) if operator == "||": return _sql_or(left, right) if left is None or right is None: return None if operator in {"+", "-", "*", "/", "%"}: left_decimal = _decimal(left) right_decimal = _decimal(right) return { "+": left_decimal + right_decimal, "-": left_decimal - right_decimal, "*": left_decimal * right_decimal, "/": left_decimal / right_decimal, "%": left_decimal % right_decimal, }[operator] if operator == "==": return left == right if operator == "!=": return left != right if operator == "<": return left < right if operator == "<=": return left <= right if operator == ">": return left > right return left >= right function = node["function"] arguments = [evaluate(argument) for argument in node["arguments"]] if function == "coalesce": return next((value for value in arguments if value is not None), None) if any(value is None for value in arguments): return None if function == "matches": return re.search(arguments[1], arguments[0]) is not None if function == "lower": return arguments[0].lower() if function == "upper": return arguments[0].upper() if function == "trim": return arguments[0].strip() if function == "length": return len(arguments[0]) if function == "date": return _timestamp(arguments[0], timezone_info).astimezone(timezone_info).date() if function == "timestamp": return _timestamp(arguments[0], timezone_info) if function == "abs": return abs(_decimal(arguments[0])) precision = int(arguments[1]) if len(arguments) == 2 else 0 return _decimal(arguments[0]).quantize( Decimal(1).scaleb(-precision), rounding=ROUND_HALF_UP ) return evaluate(ast)