Bryan Chavez
01/20/2022, 10:57 PMBryan Chavez
01/24/2022, 6:53 PMdependencies = {}
for op_config in op_configs:
def_name = op_config["op"]
alias = op_config.get("alias", def_name)
deps = op_config.get("deps", {}) or {}
is_collect = op_config.get('is_collect')
# set up op dependencies
op_deps_entry = {}
for input_name, output_data in deps.items():
if is_collect:
op_deps_entry[input_name] = MultiDependencyDefinition(
[
DependencyDefinition(
output_data['op'],
output_data.get("output", "result")
),
],
)
else:
op_deps_entry[input_name] = DependencyDefinition(
node=output_data["op"],
output=output_data.get("output", "result"),
)
dependencies[
NodeInvocation(name=def_name, alias=alias)
] = op_deps_entry
return GraphDefinition(
name=graph_name,
description=graph_description,
node_defs=op_definitions,
dependencies=dependencies,
)
alex
01/24/2022, 7:13 PMDynamicCollectDependencyDefinition
is the thing you want. I just included MultiDependencyDefinition
since it behaves in a similar way and is more well documentedBryan Chavez
01/24/2022, 9:59 PM