Shaman
63 lines
2.3 KiB
Python
63 lines
2.3 KiB
Python
"""Template loading with startup validation of graph links and mappings."""
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from .config import Settings
|
|
from .errors import AdapterError
|
|
|
|
|
|
class TemplateError(AdapterError):
|
|
"""Malformed workflow template; blocks startup."""
|
|
|
|
def __init__(self, message: str):
|
|
super().__init__(500, message, "server_error", "template_error")
|
|
|
|
|
|
def find_broken_links(graph: dict) -> list[str]:
|
|
"""References to node ids that do not exist in the graph."""
|
|
errors = []
|
|
for node_id, node in graph.items():
|
|
for field, value in node.get("inputs", {}).items():
|
|
if isinstance(value, list) and len(value) == 2 and isinstance(value[0], str):
|
|
if value[0] not in graph:
|
|
errors.append(f"node {node_id}: input '{field}' links to missing node '{value[0]}'")
|
|
return errors
|
|
|
|
|
|
def check_mapping(graph: dict, mapping: dict, name: str) -> list[str]:
|
|
"""Every mapped node id must exist with a scalar input field."""
|
|
errors = []
|
|
for key, spec in mapping.items():
|
|
node = graph.get(spec["node"])
|
|
if node is None:
|
|
errors.append(f"mapping '{name}.{key}': node '{spec['node']}' not found in template")
|
|
continue
|
|
for field in _mapped_fields(spec):
|
|
if field not in node.get("inputs", {}) or isinstance(node["inputs"][field], list):
|
|
errors.append(f"mapping '{name}.{key}': node '{spec['node']}' has no scalar input '{field}'")
|
|
return errors
|
|
|
|
|
|
def _mapped_fields(spec: dict) -> list[str]:
|
|
if "width_field" in spec:
|
|
return [spec["width_field"], spec["height_field"]]
|
|
return [spec["field"]]
|
|
|
|
|
|
def load_template(path: Path, graph_name: str, mapping: dict, mapping_name: str) -> dict:
|
|
with open(path, encoding="utf-8") as f:
|
|
graph = json.load(f)
|
|
errors = find_broken_links(graph) + check_mapping(graph, mapping[mapping_name], mapping_name)
|
|
if errors:
|
|
raise TemplateError(f"workflow template '{graph_name}' ({path}) is invalid:\n " + "\n ".join(errors))
|
|
return graph
|
|
|
|
|
|
def load_all(settings: Settings) -> tuple[dict, dict]:
|
|
mapping = settings.mapping()
|
|
create = load_template(
|
|
settings.create_template_path, "workflow_create.json", mapping, "create"
|
|
)
|
|
edit = load_template(settings.edit_template_path, "workflow_edit.json", mapping, "edit")
|
|
return create, edit
|