art-dag/celery
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix build_dag_from_recipe to use correct Node API

Browse Source 
- Use keyword arguments for Node constructor
- Pass inputs list to Node instead of calling non-existent add_edge
- Two-pass approach: create SOURCE nodes first, then resolve input
  names to content-addressed IDs for dependent nodes
- Properly set output node using resolved ID

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
gilesb
2026-01-08 23:56:53 +00:00
parent 22472e250f
commit 486cdb5d7d
1 changed files with 52 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
server.py
View File

@@ -1333,27 +1333,33 @@ def build_dag_from_recipe(yaml_config: dict, user_inputs: dict[str, str], recipe
from artdag import DAG, Node
dag = DAG()
node_map = {} # node_id -> Node
name_to_id = {} # Map YAML node names to content-addressed IDs
registry = yaml_config.get("registry", {})
assets = registry.get("assets", {})
effects = registry.get("effects", {})
dag_config = yaml_config.get("dag", {})
nodes = dag_config.get("nodes", [])
output_node = dag_config.get("output")
# First pass: create all nodes and map names to IDs
for node_def in nodes:
node_id = node_def.get("id")
node_name = node_def.get("id")
node_type = node_def.get("type")
node_config = node_def.get("config", {})
input_ids = node_def.get("inputs", [])
if node_type == "SOURCE":
if node_config.get("input"):
# Variable input - use user-provided hash
content_hash = user_inputs.get(node_id)
content_hash = user_inputs.get(node_name)
if not content_hash:
raise HTTPException(400, f"Missing input for node {node_id}")
node = Node(node_id, "SOURCE", {"content_hash": content_hash})
raise HTTPException(400, f"Missing input for node {node_name}")
node = Node(
node_type="SOURCE",
config={"content_hash": content_hash},
inputs=[],
name=node_name
)
else:
# Fixed input - use registry hash
asset_name = node_config.get("asset")
@@ -1361,24 +1367,53 @@ def build_dag_from_recipe(yaml_config: dict, user_inputs: dict[str, str], recipe
content_hash = asset_info.get("hash")
if not content_hash:
raise HTTPException(400, f"Asset {asset_name} not found in registry")
node = Node(node_id, "SOURCE", {"content_hash": content_hash})
elif node_type == "EFFECT":
node = Node(
node_type="SOURCE",
config={"content_hash": content_hash},
inputs=[],
name=node_name
)
name_to_id[node_name] = node.node_id
dag.add_node(node)
# Second pass: create nodes with inputs (now we can resolve input names to IDs)
for node_def in nodes:
node_name = node_def.get("id")
node_type = node_def.get("type")
node_config = node_def.get("config", {})
input_names = node_def.get("inputs", [])
# Skip SOURCE nodes (already added)
if node_type == "SOURCE":
continue
# Resolve input names to content-addressed IDs
input_ids = [name_to_id[name] for name in input_names if name in name_to_id]
if node_type == "EFFECT":
effect_name = node_config.get("effect")
effect_info = effects.get(effect_name, {})
effect_hash = effect_info.get("hash")
node = Node(node_id, "EFFECT", {"effect": effect_name, "effect_hash": effect_hash})
node = Node(
node_type="EFFECT",
config={"effect": effect_name, "effect_hash": effect_hash},
inputs=input_ids,
name=node_name
)
else:
node = Node(node_id, node_type, node_config)
node = Node(
node_type=node_type,
config=node_config,
inputs=input_ids,
name=node_name
)
node_map[node_id] = node
name_to_id[node_name] = node.node_id
dag.add_node(node)
# Connect edges
for node_def in nodes:
node_id = node_def.get("id")
input_ids = node_def.get("inputs", [])
for input_id in input_ids:
dag.add_edge(input_id, node_id)
# Set output node
if output_node and output_node in name_to_id:
dag.set_output(name_to_id[output_node])
return dag