193 lines
7.8 KiB
Python
193 lines
7.8 KiB
Python
from typing import Any, Dict, List, Set, Tuple
|
|
from backend.engine.sandbox import extract_variable_names, FormulaSyntaxError
|
|
|
|
|
|
class DAGValidationError(Exception):
|
|
pass
|
|
|
|
|
|
def validate_intra_table_dag(table: Dict[str, Any]) -> None:
|
|
table_name = table.get("name", "unnamed_table")
|
|
archetype = table.get("archetype", "record")
|
|
fields = table.get("fields", [])
|
|
|
|
if archetype in ("reference", "collection"):
|
|
return
|
|
|
|
seen_columns: Set[str] = set()
|
|
ordered_columns: List[str] = []
|
|
|
|
for idx, field in enumerate(fields):
|
|
name = field.get("name")
|
|
if not name:
|
|
raise DAGValidationError(f"Table '{table_name}': Column at index {idx} missing 'name'.")
|
|
if name in seen_columns:
|
|
raise DAGValidationError(f"Table '{table_name}': Duplicate column name '{name}'.")
|
|
|
|
role = field.get("role", "randomized")
|
|
|
|
if archetype == "record" and role == "aggregated":
|
|
raise DAGValidationError(
|
|
f"Table '{table_name}' is 'record' archetype but column '{name}' has forbidden role 'aggregated'."
|
|
)
|
|
|
|
config = field.get("config") or {}
|
|
|
|
if role == "function":
|
|
expr = config.get("expression", "")
|
|
if not expr:
|
|
raise DAGValidationError(f"Table '{table_name}': Function column '{name}' has empty expression.")
|
|
try:
|
|
referenced = extract_variable_names(expr)
|
|
except FormulaSyntaxError as e:
|
|
raise DAGValidationError(f"Table '{table_name}', Column '{name}': {e}") from e
|
|
|
|
for var in referenced:
|
|
if var not in seen_columns:
|
|
raise DAGValidationError(
|
|
f"Table '{table_name}': Column '{name}' references '{var}', which is not defined in upstream columns."
|
|
)
|
|
|
|
elif role == "reference":
|
|
source_field = config.get("source_field")
|
|
if not source_field:
|
|
raise DAGValidationError(f"Table '{table_name}': Reference column '{name}' missing 'source_field'.")
|
|
if source_field not in seen_columns:
|
|
raise DAGValidationError(
|
|
f"Table '{table_name}': Reference column '{name}' references source '{source_field}' not in upstream columns."
|
|
)
|
|
|
|
seen_columns.add(name)
|
|
ordered_columns.append(name)
|
|
|
|
exclude_if = table.get("exclude_if")
|
|
if exclude_if and exclude_if.strip():
|
|
try:
|
|
exclude_vars = extract_variable_names(exclude_if)
|
|
except FormulaSyntaxError as e:
|
|
raise DAGValidationError(f"Table '{table_name}', exclude_if: {e}") from e
|
|
|
|
for var in exclude_vars:
|
|
if var not in seen_columns:
|
|
raise DAGValidationError(
|
|
f"Table '{table_name}': exclude_if references '{var}' which is not a column in this table."
|
|
)
|
|
|
|
|
|
def validate_inter_table_dag(tables: List[Dict[str, Any]]) -> None:
|
|
table_map: Dict[str, Dict[str, Any]] = {}
|
|
for t in tables:
|
|
t_id = t.get("id") or t.get("name")
|
|
t_name = t.get("name")
|
|
table_map[t_id] = t
|
|
if t_name:
|
|
table_map[t_name] = t
|
|
|
|
graph: Dict[str, Set[str]] = {t.get("name", ""): set() for t in tables}
|
|
|
|
for t in tables:
|
|
t_name = t.get("name", "")
|
|
if t.get("archetype") in ("reference", "collection"):
|
|
continue
|
|
|
|
# Draft tables are work in progress; only validate outbound links of tables marked valid
|
|
if t.get("status") != "valid":
|
|
continue
|
|
|
|
for field in t.get("fields", []):
|
|
config = field.get("config") or {}
|
|
role = field.get("role")
|
|
col_type = field.get("type")
|
|
|
|
if role == "reference":
|
|
ref_table_name = config.get("ref_table")
|
|
if not ref_table_name:
|
|
raise DAGValidationError(f"Table '{t_name}', Column '{field.get('name')}': missing 'ref_table'.")
|
|
|
|
ref_target = table_map.get(ref_table_name)
|
|
if not ref_target:
|
|
raise DAGValidationError(
|
|
f"Table '{t_name}': References table '{ref_table_name}' which does not exist in this project."
|
|
)
|
|
|
|
if ref_target.get("archetype") != "reference":
|
|
raise DAGValidationError(
|
|
f"Table '{t_name}': References table '{ref_table_name}' with archetype '{ref_target.get('archetype')}'. Only 'reference' archetype tables can be referenced."
|
|
)
|
|
|
|
if ref_target.get("status") != "valid":
|
|
raise DAGValidationError(
|
|
f"Table '{t_name}': References table '{ref_table_name}' which is in '{ref_target.get('status')}' status (must be 'valid')."
|
|
)
|
|
|
|
ref_cols = ref_target.get("columns", [])
|
|
key_col = config.get("key_col")
|
|
val_col = config.get("val_col")
|
|
if key_col not in ref_cols:
|
|
raise DAGValidationError(
|
|
f"Table '{t_name}': key_col '{key_col}' not found in reference table '{ref_table_name}' columns {ref_cols}."
|
|
)
|
|
if val_col not in ref_cols:
|
|
raise DAGValidationError(
|
|
f"Table '{t_name}': val_col '{val_col}' not found in reference table '{ref_table_name}' columns {ref_cols}."
|
|
)
|
|
|
|
graph[t_name].add(ref_table_name)
|
|
|
|
elif col_type == "collection" or "collection_table" in config:
|
|
coll_table_name = config.get("collection_table")
|
|
if not coll_table_name:
|
|
raise DAGValidationError(f"Table '{t_name}', Column '{field.get('name')}': missing 'collection_table'.")
|
|
|
|
coll_target = table_map.get(coll_table_name)
|
|
if not coll_target:
|
|
raise DAGValidationError(
|
|
f"Table '{t_name}': References collection '{coll_table_name}' which does not exist in this project."
|
|
)
|
|
|
|
if coll_target.get("archetype") != "collection":
|
|
raise DAGValidationError(
|
|
f"Table '{t_name}': References table '{coll_table_name}' with archetype '{coll_target.get('archetype')}'. Must be 'collection' archetype."
|
|
)
|
|
|
|
if coll_target.get("status") != "valid":
|
|
raise DAGValidationError(
|
|
f"Table '{t_name}': References collection '{coll_table_name}' which is in '{coll_target.get('status')}' status (must be 'valid')."
|
|
)
|
|
|
|
graph[t_name].add(coll_table_name)
|
|
|
|
# Cycle check
|
|
visited: Dict[str, int] = {} # 0: unvisited, 1: visiting, 2: visited
|
|
|
|
def has_cycle(node: str) -> bool:
|
|
visited[node] = 1
|
|
for neighbor in graph.get(node, set()):
|
|
state = visited.get(neighbor, 0)
|
|
if state == 1:
|
|
return True
|
|
if state == 0 and has_cycle(neighbor):
|
|
return True
|
|
visited[node] = 2
|
|
return False
|
|
|
|
for node in graph:
|
|
if visited.get(node, 0) == 0:
|
|
if has_cycle(node):
|
|
raise DAGValidationError(f"Circular reference detected involving table '{node}'.")
|
|
|
|
|
|
def get_referencing_tables(target_table_name: str, all_tables: List[Dict[str, Any]]) -> List[str]:
|
|
referencers = []
|
|
for t in all_tables:
|
|
if t.get("name") == target_table_name:
|
|
continue
|
|
for field in t.get("fields", []):
|
|
cfg = field.get("config") or {}
|
|
ref_t = cfg.get("ref_table")
|
|
coll_t = cfg.get("collection_table")
|
|
if ref_t == target_table_name or coll_t == target_table_name:
|
|
referencers.append(t.get("name", "unnamed"))
|
|
break
|
|
return referencers
|