Source code for dataprocessor.pipes.add_node

# coding=utf-8

import os.path
from .. import nodes
from .. import utility


[docs]def add_node(node_list, path=".", node_type="run", children=[], name=None, parents=[".."], strategy="raise"): """Add node. A node is added to the node_list. Parameters ---------- path : str, optional path of node (default=".") node_type : str, optional {"run", "project"} type of node (default="run") name : str, optional children : list, optional parents : list, optional strategy : str, optional determines how to deal with conflict, i.e. there alread exists another node whose path is same as new one. See docstring of nodes.add. Returns ------- node_list """ path = utility.path_expand(path) if isinstance(children, str): children = [children] if isinstance(parents, str): parents = [parents] if not name: name = os.path.basename(path) node = nodes.normalize({ "path": path, "type": node_type, "children": [utility.path_expand(c_path) for c_path in children], "parents": [utility.path_expand(c_path) for c_path in parents], "name": name }) nodes.add(node_list, node, strategy=strategy) return node_list
[docs]def register(pipes_dics): pipes_dics["add_node"] = { "func": add_node, "args": [], "kwds": [ ("path", {"help": "path of new node"}), ("name", {"help": "name of new node"}), ("node_type", { "help": "type of new node", "choices": nodes.node_types, }), ("children", {"help": "children paths of new node"}), ("parents", {"help": "paths parents of new node"}), ("strategy", { "help": "strategy for resolving conflict", "choices": ["raise", "update", "modest_update", "replace"], }), ], "desc": "Add node to node_list." }