def _expand_sub_pipelines(self, validate=True):
        log.info("Expand | Checking nodes for sub-pipelines")
        check_fanout = True
        for node in self.topological_order():
            log.debug("Expand | Checking %s for sub-pipeline", node)

            # setup and render the subpipe node. We
            # do this so that local variables used in the pipeline
            # are rendered properly and the values are set accordingly
            #if hasattr(node._tool, 'pipeline'):
            node._tool.setup()
            # reapply the pipeline profile so it precedes the tool profile
            if node._pipeline_profile:
                node._pipeline_profile.update(node._job, overwrite=False)
                node._job.update(node._pipeline_profile)
            # make the nodes options absolute (Issue #38)
            node._tool.options.make_absolute(node._job.working_dir)
            _render_nodes(self, [node])
            node._tool.options.make_absolute(node._job.working_dir)
            sub_pipe = node._tool.pipeline()
            if sub_pipe is None:
                continue
            # validate the sub-pipeline
            self._validate_node(node, silent=not validate)

            # merge the nodes jobs with the sub-pipeline nodes
            for sub_node in sub_pipe.nodes():
                node._job.apply_to_node(sub_node)
                sub_node._job.merge(node._job)

            # render and apply the nodes name as pipeline
            # name
            node._name = node.job._render_name()
            sub_pipe.name(node.name)
            node._pipeline_name = self._pipeline_name

            log.info("Expand | Expanding sub-pipeline from node %s", node)
            if sub_pipe.excludes:
                self.excludes.extend(sub_pipe.excludes)
            check_fanout = sub_pipe.expand(validate=validate, _find_dup=False,
                                           _check_fanout=check_fanout)

            # apply pipeline profile
            node._job.apply_to_pipeline(sub_pipe)
            _render_jobs(sub_pipe, list(sub_pipe.nodes()))

            # find all nodes in the sub_pipeline
            # with no incoming edges and connect
            # them to the current nodes incoming nodes
            no_incoming = [n for n in sub_pipe.nodes()
                           if len(list(n.incoming())) == 0]
            no_outgoing = [n for n in sub_pipe.nodes()
                           if len(list(n.outgoing())) == 0]
            # add the sub_pipe
            log.info("Expand | Adding %d nodes from sub-pipeline",
                     len(sub_pipe))
            for sub_node in sub_pipe.topological_order():
                log.debug("Expand | Adding sub-pipeline node %s", sub_node)
                self.add(sub_node)
            self._edges = self._edges.union(sub_pipe._edges)

            for inedge in node.incoming():
                for target in no_incoming:
                    log.debug("Expansion | add edge dependency on "
                              "no-incoming edge %s->%s",
                              inedge._source, target)
                    self.add_edge(inedge._source, target)

            for outedge in node.outgoing():
                for source in no_outgoing:
                    log.debug("Expansion | add edge dependency on "
                              "no-outgoing edge %s->%s",
                              source, outedge._target)
                    self.add_edge(source, outedge._target)

            # establish links between resolved nodes and the current pipeline
            # where before, the node was linked against a pipeline option.
            #
            # we look for both incoming and outgoing edges of the old node
            # and check their links. If we find a link where source/target
            # option is in one of the new sub_nodes _pipeline_options, we
            # reestablish the link between the options, now linking between
            # the nodes
            self._expand_subpipe_resolve_outgoing(node, sub_pipe)
            self._expand_subpipe_resolve_incoming(node, sub_pipe)
            # setup the node and render values before we remove the node
            node._tool.setup()
            # reapply the pipeline profile so it precedes the tool profile
            if node._pipeline_profile:
                node._pipeline_profile.update(node._job, overwrite=False)
                node._job.update(node._pipeline_profile)
            _create_render_context(self, node._tool, node)
            #_render_nodes(self, [node])

            self.remove(node, remove_links=False)
            self._cleanup_nodes.extend(sub_pipe._cleanup_nodes)
Called on completion of an update.
dataset/ETHPy150Open thasso/pyjip/jip/pipelines.py/Pipeline._expand_sub_pipelines--crossover with--dataset/ETHPy150Open viewfinderco/viewfinder/backend/db/schema.py/_Value.OnUpdate