def __init__()

in lib/ramble/ramble/graphs.py [0:0]

• 60 lines of code
• 25 McCabe index (conditional complexity)

    def __init__(self, exec_order, executables, builtin_objects, builtin_groups, obj_inst):
        """Construct a new ExecutableGraph

        Executable graphs have node attributes that are either of type
        CommandExecutable, or are a function pointer to a builtin.

        Args:
            exec_order (list(str)): List of executable names in execution order
            executables (dict): Dictionary of executable definitions.
                                Keys are executable names, values are CommandExecutables
            builtin_objects (list(object)): List of objects to associate with each builtin
                                            group (in order)
            builtins (list(dict)): List of dictionaries containing definitions of builtins.
                                   Keys are names values are configurations of builtins.
            modifier_builtins (dict): Dictionary containing definitions of modifier builtins.
                                      Keys are names values are configurations of builtins.
                                      Modifier builtins are inserted between application builtins
                                      and executables.
            obj_inst (object): Object instance to extract attributes from (when necessary)
        """
        super().__init__(obj_inst)
        self._builtin_dependencies = defaultdict(list)
        # Mapping from shorter_name -> fully qualified names
        self._builtin_aliases = defaultdict(list)

        # Define nodes for executable
        for exec_name, cmd_exec in executables.items():
            exec_node = ramble.util.graph.GraphNode(exec_name, cmd_exec, obj_inst=obj_inst)
            self.node_definitions[exec_name] = exec_node
            if exec_name in exec_order:
                super().update_graph(exec_node)

        # Define nodes for builtins
        for builtin_obj, builtins in zip(builtin_objects, builtin_groups):
            for builtin, blt_conf in builtins.items():
                self._builtin_dependencies[builtin] = blt_conf["depends_on"].copy()
                blt_func = getattr(builtin_obj, blt_conf["name"])
                exec_node = ramble.util.graph.GraphNode(
                    builtin, attribute=blt_func, obj_inst=builtin_obj
                )
                self.node_definitions[builtin] = exec_node
                self._build_builtin_aliases(builtin)
            for builtin, blt_conf in builtins.items():
                dependents = blt_conf["dependents"].copy()
                for dependent in dependents:
                    self._builtin_dependencies[dependent].append(builtin)

        dep_exec = None
        for exec_name in exec_order:
            if dep_exec is not None:
                exec_node = self.node_definitions[exec_name]
                dep_node = self.node_definitions[dep_exec]
                super().update_graph(exec_node, [dep_node])
            dep_exec = exec_name

        head_node = None
        tail_node = None
        for node in self.walk():
            if head_node is None:
                head_node = node
            tail_node = node

        # Add (missing) required builtins
        for builtins in builtin_groups:
            for builtin, blt_conf in builtins.items():
                if blt_conf["required"] and self.get_node(builtin) is None:
                    blt_node = self.node_definitions[builtin]
                    super().update_graph(blt_node)

                    if blt_conf["injection_method"] == "prepend":
                        if head_node is not None:
                            super().update_graph(head_node, [blt_node])
                    elif blt_conf["injection_method"] == "append":
                        if tail_node is not None:
                            super().update_graph(blt_node, [tail_node])

                    if blt_conf["depends_on"]:
                        deps = []
                        for dep in blt_conf["depends_on"]:
                            dep_node = self._resolve_builtin_node(dep)
                            super().update_graph(dep_node)
                            deps.append(dep_node)

                        exec_node = self.node_definitions[builtin]
                        super().update_graph(exec_node, deps)

                    if blt_conf["dependents"]:
                        exec_node = self.node_definitions[builtin]
                        super().update_graph(exec_node)
                        for dependent in blt_conf["dependents"]:
                            dependent_node = self._resolve_builtin_node(dependent)
                            super().update_graph(dependent_node, [exec_node])