"""
Form Logic Engine — Phase 3
Server-side validation, conditional logic, and field calculations.
Field config schema (extended from Phase 2):
{
"key": "budget",
"type": "number",
"label": "Budget",
"required": true,
"placeholder": "Enter your budget",
"options": [], # select/radio/checkbox groups
"default": "",
"step": "any", # number step
"min": 0, "max": 10000, # number range
"minLength": 2, "maxLength": 200, # text/email/textarea length
"pattern": "^[A-Z]{2}\\\\d{4}$", # custom regex
"errorMessage": "Must be 2 letters + 4 digits",
"condition": { # conditional visibility (single or group)
"all": [ # AND group (Phase 3)
{ "field": "service", "operator": "eq", "value": "consulting" },
{ "field": "tier", "operator": "in", "value": "pro,enterprise" }
],
# or "any": [...] for OR group
# or legacy single: { "field": "x", "operator": "eq", "value": "y" }
},
"conditional_required": { # conditional required (Phase 3)
"field": "service",
"operator": "eq",
"value": "premium"
},
"conditional_default": { # conditional default value (Phase 3)
"condition": { "field": "service", "operator": "eq", "value": "default_svc" },
"value": "auto-filled-value"
},
"conditional_validation": [ # conditional validation rules (Phase 3)
{
"condition": { "field": "country", "operator": "eq", "value": "US" },
"pattern": "^[A-Z]{2}\\d{5}$",
"errorMessage": "Must be US zip code"
}
],
"calculation": { # computed fields (type: computed)
"formula": "{{quantity}} * {{unit_price}}",
"round": 2
},
"step": 1 # multi-step grouping
}
"""
import math
import re
from typing import Any
class ValidationError:
"""A single validation failure on a field."""
def __init__(self, field: str, message: str, rule: str | None = None):
self.field = field
self.message = message
self.rule = rule
def to_dict(self) -> dict:
d = {"field": self.field, "message": self.message}
if self.rule:
d["rule"] = self.rule
return d
def validate_field(
field_def: dict, value: Any, all_data: dict, overrides: dict | None = None
) -> ValidationError | None:
"""Validate a single field value against its definition.
Returns ValidationError on failure, None on success.
Args:
field_def: Field definition dict
value: User-submitted value
all_data: Full submission data (for context)
overrides: Optional dict with validation overrides from conditional rules.
Keys: required, pattern, minLength, maxLength, min, max, errorMessage
Checks applied (in order):
1. Required
2. Pattern (regex)
3. Min/Max length (text/email/textarea)
4. Min/Max value (number)
5. Range for select/in options
"""
key = field_def.get("key") or field_def.get("name", "")
# Normalize: ensure both key and name are set
if not field_def.get("key"):
field_def["key"] = key
if not field_def.get("name"):
field_def["name"] = key
ftype = field_def.get("type", "text")
# Merge overrides — overrides take precedence over base config
is_required = field_def.get("required", False)
if overrides and overrides.get("required") is not None:
is_required = bool(overrides["required"])
pattern = field_def.get("pattern")
error_message = field_def.get("errorMessage")
min_len = field_def.get("minLength")
max_len = field_def.get("maxLength")
min_val = field_def.get("min")
max_val = field_def.get("max")
if overrides:
if overrides.get("pattern") is not None:
pattern = overrides["pattern"]
if overrides.get("errorMessage") is not None:
error_message = overrides["errorMessage"]
if overrides.get("minLength") is not None:
min_len = overrides["minLength"]
if overrides.get("maxLength") is not None:
max_len = overrides["maxLength"]
if overrides.get("min") is not None:
min_val = overrides["min"]
if overrides.get("max") is not None:
max_val = overrides["max"]
# Required check
if is_required:
if value is None or value == "":
return ValidationError(
key,
error_message or f"{field_def.get('label', key)} is required.",
"required",
)
# Skip further validation if empty and optional
if value is None or value == "":
return None
# Checkbox: only "required" applies (checked = "1"/"true")
if ftype == "checkbox":
return None
# Pattern (regex) — applies to text/email/phone/textarea
# Check both direct field pattern and validation dict
validation = field_def.get("validation") or {}
if not pattern and validation.get("type") == "pattern":
pattern = validation.get("pattern")
if pattern:
try:
if not re.search(pattern, str(value)):
return ValidationError(
key,
error_message or "Value does not match the expected format.",
"pattern",
)
except re.error:
pass # Bad regex in config — skip silently
# Email format validation — check validation.type == "email" or type == "email"
if ftype == "email" or validation.get("type") == "email":
if not re.match(r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$", str(value)):
return ValidationError(
key,
error_message or "Please enter a valid email address.",
"email",
)
# Min/Max length — text/email/textarea/phone
if ftype in ("text", "email", "textarea", "phone"):
str_value = str(value)
if min_len is not None and len(str_value) < int(min_len):
return ValidationError(
key,
error_message or f"Minimum {min_len} characters required.",
"minLength",
)
if max_len is not None and len(str_value) > int(max_len):
return ValidationError(
key,
error_message or f"Maximum {max_len} characters allowed.",
"maxLength",
)
# Min/Max value — number
if ftype == "number":
try:
num_value = float(value)
if min_val is not None and num_value < float(min_val):
return ValidationError(
key,
error_message or f"Must be at least {min_val}.",
"min",
)
if max_val is not None and num_value > float(max_val):
return ValidationError(
key,
error_message or f"Must be at most {max_val}.",
"max",
)
except (ValueError, TypeError):
return ValidationError(
key,
error_message or "Please enter a valid number.",
"number",
)
# Options validation — select/radio must be in list
if ftype in ("select", "radio") and field_def.get("options"):
options = field_def["options"]
if str(value) not in [str(o) for o in options]:
return ValidationError(
key,
error_message or "Please select a valid option.",
"options",
)
return None
def _evaluate_single_condition(condition: dict, all_data: dict) -> bool:
"""Evaluate a single condition {field, operator, value} against data.
This is the core operator logic — unchanged from Phase 2.
Returns True if condition is met (field should be SHOWN), False to HIDE.
"""
target_field = condition.get("field", "")
operator = condition.get("operator", "eq")
target_value = all_data.get(target_field, "")
# If target field doesn't exist in data, hide dependent field
if target_value is None:
return False
target_value = str(target_value).strip()
compare_value = str(condition.get("value", "")).strip()
# Boolean operators
if operator == "filled":
return bool(target_value)
if operator == "empty":
return not bool(target_value)
# Numeric comparisons
if operator in ("gt", "lt", "gte", "lte"):
try:
t, c = float(target_value), float(compare_value)
except (ValueError, TypeError):
return False
if operator == "gt":
return t > c
if operator == "lt":
return t < c
if operator == "gte":
return t >= c
if operator == "lte":
return t <= c
# String comparisons
if operator == "eq":
return target_value == compare_value
if operator == "neq":
return target_value != compare_value
if operator == "contains":
return compare_value in target_value
if operator == "not_contains":
return compare_value not in target_value
# Set membership
if operator == "in":
values = [v.strip() for v in compare_value.split(",")]
return target_value in values
if operator == "not_in":
values = [v.strip() for v in compare_value.split(",")]
return target_value not in values
return True # Unknown operator = visible (fail open)
def evaluate_condition(field_def: dict, all_data: dict) -> bool:
"""Check if a field should be visible based on its condition.
Supports:
- Legacy single condition: {field, operator, value}
- AND group: {all: [cond1, cond2, ...]}
- OR group: {any: [cond1, cond2, ...]}
- Nested groups: all can contain any which can contain all (recursive)
Returns True if the field should be SHOWN, False if it should be HIDDEN.
If no condition is defined, the field is always visible (True).
"""
condition = field_def.get("condition")
if not condition:
return True # No condition = always visible
# AND group — all sub-conditions must be true
if "all" in condition:
return all(evaluate_condition({"condition": sub}, all_data) for sub in condition["all"])
# OR group — at least one sub-condition must be true
if "any" in condition:
return any(evaluate_condition({"condition": sub}, all_data) for sub in condition["any"])
# Legacy single condition — has 'field' key
if "field" in condition:
return _evaluate_single_condition(condition, all_data)
return True # Unknown format = visible (fail open)
def _get_field_state(field_def: dict, data: dict, visible_fields: dict[str, bool]) -> dict:
"""Compute the full state for a single field.
Returns:
{
"visible": bool,
"required": bool,
"default": str|None,
"validation_overrides": dict,
}
"""
key = field_def.get("key") or field_def.get("name", "")
# Visibility
visible = evaluate_condition(field_def, data)
# Also check if any referenced condition field is hidden — if so, this field is hidden too
# (cascade: if B depends on A, and A is hidden, B is hidden)
if visible:
for ref_field in _extract_referenced_fields(field_def):
if ref_field and not visible_fields.get(ref_field, True):
visible = False
break
# Required — base + conditional override
required = bool(field_def.get("required", False))
cond_required = field_def.get("conditional_required")
if cond_required:
if _evaluate_single_condition(cond_required, data):
required = True
elif not field_def.get("required", False):
# If base is not required and conditional doesn't fire, not required
required = False
# Default — base + conditional override
default = field_def.get("default", "")
cond_default = field_def.get("conditional_default")
if cond_default:
cond_condition = cond_default.get("condition", {})
if _evaluate_single_condition(cond_condition, data):
default = cond_default.get("value", "")
# Validation overrides — conditional_validation array
validation_overrides = {}
cond_validations = field_def.get("conditional_validation", [])
for cv in cond_validations:
cond_condition = cv.get("condition", {})
if _evaluate_single_condition(cond_condition, data):
validation_overrides = {
k: cv[k] for k in ("pattern", "errorMessage", "minLength", "maxLength", "min", "max") if k in cv
}
break # First matching rule wins
return {
"visible": visible,
"required": required,
"default": default,
"validation_overrides": validation_overrides,
}
def _extract_referenced_fields(field_def: dict) -> list[str]:
"""Extract all field names referenced by conditions in a field definition."""
refs = []
condition = field_def.get("condition")
if not condition:
return refs
if isinstance(condition, dict):
if "field" in condition:
refs.append(condition["field"])
for group_key in ("all", "any"):
if group_key in condition:
for sub in condition[group_key]:
if isinstance(sub, dict) and "field" in sub:
refs.append(sub["field"])
# Recurse for nested groups
refs.extend(_extract_referenced_fields({"condition": sub}))
return refs
def evaluate_all_conditions(field_config: list | dict, data: dict, max_iterations: int = 3) -> dict[str, dict]:
"""Evaluate all field conditions with cascade support.
Returns:
{
field_key: {
"visible": bool,
"required": bool,
"default": str|None,
"validation_overrides": dict,
}
}
Cascade: if A hides B, and B's value is used by C's condition, C may also hide.
We iterate up to max_iterations times until the state stabilizes.
"""
# Accept either a list of fields or a dict with 'fields' key
if isinstance(field_config, dict):
fields = field_config.get("fields", [])
else:
fields = field_config
states = {}
for iteration in range(max_iterations):
new_states = {}
visible_fields = {k: v["visible"] for k, v in states.items()}
for field_def in fields:
key = field_def.get("key") or field_def.get("name", "")
new_states[key] = _get_field_state(field_def, data, visible_fields)
# Check if state stabilized
if new_states == states:
break
states = new_states
return states
def resolve_formula(formula: str, data: dict) -> Any:
"""Resolve a template formula like " {{a}} * {{b}}" against data.
Supports arithmetic: +, -, *, /, %, parentheses.
Returns the computed result or None if formula can't be evaluated.
"""
import ast
import operator
from simpleeval import SimpleEval, simple_eval
# Secure operator map: standard arithmetic + ^ as power alias
# Replaces the original operators={"^": pow} which broke all defaults
_SAFE_OPS = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.FloorDiv: operator.floordiv,
ast.Mod: operator.mod,
ast.Pow: operator.pow,
ast.USub: operator.neg,
ast.UAdd: operator.pos,
ast.BitXor: operator.pow, # ^ as power instead of XOR
}
_SAFE_ATTRS = {
"__add__",
"__sub__",
"__mul__",
"__truediv__",
"__floordiv__",
"__mod__",
"__pow__",
"__neg__",
"__pos__",
}
# Replace {{key}} with values
resolved = formula
for key, value in data.items():
resolved = resolved.replace("{{" + key + "}}", str(value) if value is not None else "0")
# Check if any unresolved {{key}} remain
if "{{" in resolved:
return None
# Evaluate as arithmetic expression using simpleeval
# names={} blocks variable/attribute access — only numeric literals allowed
# allowed_attrs restricts to numeric dunder methods only
try:
result = simple_eval(
resolved,
operators=_SAFE_OPS,
names={},
allowed_attrs=_SAFE_ATTRS,
)
# Handle division by zero
if isinstance(result, float) and (result == float("inf") or result == float("-inf") or math.isnan(result)):
return None
return result
except Exception:
pass
# If not arithmetic, return as string (for string concatenation formulas)
return resolved
def compute_field(field_def: dict, all_data: dict) -> Any:
"""Compute a field value.
Supports three methods:
1. calculation.formula — template expression (legacy)
2. computed_method: 'map' with computed_from + computed_map
3. computed_method: 'expression' with computed_expression
"""
# Method 1: calculation.formula (legacy)
calc = field_def.get("calculation")
if calc:
formula = calc.get("formula", "")
if formula:
result = resolve_formula(formula, all_data)
rounding = calc.get("round")
if rounding is not None and isinstance(result, (int, float)):
result = round(result, int(rounding))
if isinstance(result, float) and result == int(result):
result = int(result)
return result
# Method 2: computed_method: 'map'
if field_def.get("computed_method") == "map":
source = field_def.get("computed_from", "")
mapping = field_def.get("computed_map", {})
source_value = str(all_data.get(source, "")).strip()
result = mapping.get(source_value, field_def.get("default", ""))
# Coerce numeric strings
if isinstance(result, str):
try:
result = int(result)
except (ValueError, TypeError):
try:
result = float(result)
except (ValueError, TypeError):
pass
return result
# Method 3: computed_method: 'expression'
if field_def.get("computed_method") == "expression":
expr = field_def.get("computed_expression", "")
result = resolve_formula(expr, all_data)
if isinstance(result, float) and result == int(result):
result = int(result)
return result
return all_data.get(field_def.get("key", ""), field_def.get("default", ""))
def validate_submission(field_config: list | dict, data: dict) -> list[dict]:
"""Validate an entire submission against field definitions.
Returns list of ValidationError dicts. Empty list = valid.
Only validates fields that are visible per conditions.
Phase 3: Uses evaluate_all_conditions() for cascade + conditional behavior.
"""
errors = []
if not field_config:
return errors
# Normalize: accept list or dict with 'fields' key
if isinstance(field_config, dict):
fields = field_config.get("fields", [])
else:
fields = field_config
# Evaluate all conditions once (with cascade)
field_states = evaluate_all_conditions(field_config, data)
for field_def in fields:
# Normalize key/name (existing configs use 'name', new ones use 'key')
if not field_def.get("key"):
field_def["key"] = field_def.get("name", "")
if not field_def.get("name"):
field_def["name"] = field_def.get("key", "")
key = field_def["key"]
# Skip computed fields (they don't accept user input)
if field_def.get("type") == "computed":
continue
state = field_states.get(key, {})
# Check conditional visibility
if not state.get("visible", True):
continue # Hidden fields are not validated
value = data.get(key)
# Apply conditional behavior overrides
overrides = state.get("validation_overrides")
if state.get("required"):
overrides = overrides or {}
overrides["required"] = True
error = validate_field(field_def, value, data, overrides=overrides)
if error:
errors.append(error.to_dict())
return errors
def resolve_visible_fields(field_config: list | dict, data: dict) -> list[dict]:
"""Return only the fields that should be visible given current data.
Used by the SDK for conditional rendering.
Phase 3: Uses evaluate_all_conditions for cascade support.
"""
if isinstance(field_config, dict):
fields = field_config.get("fields", [])
else:
fields = field_config
field_states = evaluate_all_conditions(field_config, data)
visible = []
for field_def in fields:
key = field_def.get("key") or field_def.get("name", "")
if field_states.get(key, {}).get("visible", True):
visible.append(field_def)
return visible
def resolve_multi_steps(field_config: list | dict) -> dict[int, list[dict]]:
"""Group fields into steps.
Returns {step_number: [fields]}. Fields without a step go into step 1.
"""
if isinstance(field_config, dict):
fields = field_config.get("fields", [])
else:
fields = field_config
steps: dict[int, list[dict]] = {}
for field_def in fields:
step = int(field_def.get("step", 1))
steps.setdefault(step, []).append(field_def)
return dict(sorted(steps.items()))
def compute_all_fields(field_config: list | dict, data: dict) -> dict:
"""Compute all computed fields and merge with user data.
Fields are computed in order so earlier computed values can feed
later formulas (e.g. tax depends on subtotal).
"""
if isinstance(field_config, dict):
fields = field_config.get("fields", [])
else:
fields = field_config
result = dict(data)
for field_def in fields:
fd = dict(field_def) # avoid mutating the original
if not fd.get("key"):
fd["key"] = fd.get("name", "")
if fd.get("type") == "computed":
key = fd.get("key", "")
result[key] = compute_field(fd, result) # use result for cascading
return result
def build_config_response(site: dict, field_config: list | None) -> dict:
"""Build the form config response for the SDK.
Includes: fields, honeypot_enabled, multi-step info, step grouping.
"""
fields = field_config or []
# Normalize key/name for all fields
for f in fields:
if not f.get("key"):
f["key"] = f.get("name", "")
if not f.get("name"):
f["name"] = f.get("key", "")
response = {
"fields": fields,
"honeypot_enabled": site.get("honeypot_enabled", True),
"name": site.get("name", "Form"),
}
# Multi-step detection
steps = resolve_multi_steps(fields)
if len(steps) > 1:
response["multi_step"] = True
response["steps"] = {str(k): v for k, v in steps.items()}
return response