diff --git a/pyproject.toml b/pyproject.toml
index 1145d4c..a2f9e2f 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -53,7 +53,8 @@ dependencies = [
     "kaleido==0.2.1",
     "google-cloud-storage",
     "plotly~=5.11.0",
-    "inquirer~=3.1.2"
+    "inquirer~=3.1.2",
+    "networkx~=3.0"
 ]
 
 [project.optional-dependencies]
diff --git a/tenet/data/plugin.py b/tenet/data/plugin.py
new file mode 100644
index 0000000..c1b8c98
--- /dev/null
+++ b/tenet/data/plugin.py
@@ -0,0 +1,58 @@
+from dataclasses import dataclass, field
+
+
+@dataclass
+class Sources:
+    assets: dict = field(default_factory=lambda: {})
+    modified: dict = field(default_factory=lambda: {})
+
+    def is_init(self):
+        return all(self.modified.values())
+
+    def __len__(self):
+        return len(self.assets)
+
+    def __getitem__(self, key: str):
+        return self.assets[key]
+
+    def __setitem__(self, key: str, value):
+        self.modified[key] = True
+        self.assets[key] = value
+
+    def __iter__(self):
+        return iter(self.assets)
+
+    def keys(self):
+        return self.assets.keys()
+
+    def items(self):
+        return self.assets.items()
+
+    def values(self):
+        return self.assets.values()
+
+
+@dataclass
+class Sinks:
+    assets: dict = field(default_factory=lambda: {})
+
+    def __len__(self):
+        return len(self.assets)
+
+    def __getitem__(self, key: str):
+        return self.assets[key]
+
+    def __setitem__(self, key: str, value):
+        self.assets[key] = value
+
+    def __iter__(self):
+        return iter(self.assets)
+
+    def keys(self):
+        return self.assets.keys()
+
+    def items(self):
+        return self.assets.items()
+
+    def values(self):
+        return self.assets.values()
diff --git a/tenet/data/schema.py b/tenet/data/schema.py
index 149df10..1913be9 100644
--- a/tenet/data/schema.py
+++ b/tenet/data/schema.py
@@ -1,40 +1,23 @@
-import re
+import networkx as nx
 
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Dict, Any, Tuple, Callable
-from schema import Schema, And, Use, Optional, Or
+from typing import Dict, Any, Tuple, Callable, List, Union, OrderedDict
 
-from tenet.core.exc import TenetError
-from tenet.utils.misc import random_id
-
-_plugin = Schema(
-    And({'name': str, Optional('kwargs', default={}): dict},
-        Use(lambda p: Plugin(**p))))
+from schema import Schema, And, Optional, Or
 
-_edges = Schema(
-    And([{str: {'node': str, Optional('sinks', default={}): dict, Optional('sources', default=[]): list,
-                Optional('placeholders', default={}): dict, Optional('kwargs', default={}): dict}}],
-        Use(lambda els: {k: Edge(name=k, sinks=v['sinks'], sources=v['sources'], kwargs=v['kwargs'], node=v['node'],
-                                 placeholders={t: Placeholder(tag=t, value=p, node=k) for t, p in v['placeholders'].items()}) for el in els for k, v in el.items()}))
-)
-
-_layers = Schema(And({str: _edges}, Use(lambda l: {k: Layer(edges=v) for k, v in l.items()})))
+from tenet.utils.misc import random_id
 
-_nodes = Schema(
-    And(
-        [{'plugin': _plugin}],
-        Use(lambda n: {v.name: v for el in n for k, v in el.items()})
-    )
-)
 
-_workflows = Schema(And({str: list}, Use(lambda w: {k: v for k, v in w.items()})))
+_nodes = Schema(And([{str: {'plugin': str, Optional('kwargs', default={}): dict}}]))
+_edges = Schema(And([{str: {'at': str, 'to': str, Optional('links', default={}): dict}}]))
+_links = Schema(And({str: {str: {str: [str]}}}))
 
 
 @dataclass
 class Connector:
     """
-        Data class representing edge connections
+        Data class representing node connections
     """
     source: str
     sink: str
@@ -42,32 +25,20 @@ class Connector:
     links: Dict[str, str] = field(default_factory=lambda: {})
     attrs: Dict[str, Any] = field(default_factory=lambda: {})
 
-    def map_placeholders(self):
-        """
-            Looks for placeholders in the attributes and returns them with their associated values.
-        """
-        matches = {}
-
-        for source_attr, sink_attr in self.links.items():
-            if re.search("(p\d+)", sink_attr):
-                matches[sink_attr] = self.attrs[source_attr]
-
-        return matches
-
     def __getitem__(self, key: str):
         for source_attr, sink_attr in self.links.items():
-
             if key == sink_attr:
-                return self.attrs[source_attr]
+                return self.attrs[sink_attr]
             # in case the key is used inside the same plugin
             if key == source_attr:
                 return self.attrs[source_attr]
 
-        raise ValueError(f'Key {key} for sink {self.sink} not found in links. Set it in {self.source} source.')
+        raise ValueError(f"Key '{key}' for sink '{self.sink}' not found in links. Set it in '{self.source}' source.")
 
     def __setitem__(self, key: str, value: Any):
-        if key in self.attrs:
-            self.attrs[key] = value
+        # TODO: check if 'if' is necessary
+        # if key in self.attrs:
+        self.attrs[key] = value
 
     def is_init(self) -> bool:
         """
@@ -94,70 +65,181 @@ def match(self, attr: str, kind: Any = None):
 
 
 @dataclass
-class Edge:
+class ContainerCommand:
+    """
+        Data object representing the container command.
+    """
+    org: str
+    parsed: str = None
+    skip: bool = True
+    parse_fn: Callable = None
+    tag: str = None
+
+    def __str__(self):
+        return self.parsed if self.parsed else self.org
+
+
+@dataclass
+class Node:
     """
         Data object representing connections between nodes
     """
     name: str
-    node: str
-    sinks: dict = field(default_factory=lambda: {})
-    sources: list = field(default_factory=lambda: [])
-    placeholders: dict = field(default_factory=lambda: {})
+    plugin: str
+    layer: str = None
     kwargs: dict = field(default_factory=lambda: {})
 
-    @property
-    def connectors(self):
-        cons = {}
-        cons.update(self.sinks)
-        # cons.update({l: k for k, v in self.sources.items() for l in v})
-
-        return cons
-
 
 @dataclass
-class Plugin:
+class Edge:
     """
-        Code extensions
+        Data object representing connections between nodes
     """
     name: str
-    kwargs: dict = field(default_factory=lambda: {})
+    at: Node
+    to: Node
+    links: dict = field(default_factory=lambda: {})
+
+    def to_list(self) -> list:
+        return [self.at, self.to]
 
 
 @dataclass
-class Placeholder:
-    tag: str
-    node: str
-    value: Any
+class Layer:
+    name: str
+    edges: Dict[str, Edge]
 
-    def __str__(self):
-        return f"{self.node} - {self.tag}={self.value}"
+    def to_list(self,):
+        return [(e.at, e.to) for e in self.edges.values()]
+
+    def get_nodes(self):
+        return [n for e in self.edges.values() for n in [e.at, e.to]]
+
+    def set_nodes_layer(self, at_node: bool = True, to_node: bool = True):
+        for edge in self.edges.values():
+            if at_node and not edge.at.layer:
+                edge.at.layer = self.name
+            if to_node and not edge.to.layer:
+                edge.to.layer = self.name
 
 
 @dataclass
-class ContainerCommand:
-    """
-        Data object representing the container command.
-    """
-    org: str
-    parsed: str = None
-    skip: bool = True
-    placeholders: Dict[str, Placeholder] = field(default_factory=lambda: {})
-    parse_fn: Callable = None
-    tag: str = None
+class ParallelLayer:
+    name: str
+    tracks: Dict[str, List[Edge]]
 
-    def get_placeholders(self):
-        return {t: p.value for t, p in self.placeholders.items()}
+    def to_list(self) -> list:
+        return [(e.at, e.to) for t, v in self.tracks.items() for e in v]
+
+    def set_nodes_layer(self, at_node: bool = True, to_node: bool = False):
+        for track in self.tracks.values():
+            for edge in track:
+                # init layer for 'at nodes' only
+                if to_node and not edge.to.layer:
+                    edge.to.layer = self.name
+                if at_node and not edge.at.layer:
+                    edge.at.layer = self.name
+
+    def get_nodes(self):
+        nodes = []
+        for t, v in self.tracks.items():
+            for e in v:
+                if e.at not in nodes:
+                    nodes.append(e.at)
+                if e.to not in nodes:
+                    nodes.append(e.to)
+
+        return nodes
+
+
+@dataclass
+class DAGPath:
+    nodes: List[Union[Node, ParallelLayer]]
+
+    def flatten(self) -> List[Node]:
+        res = []
+        last_node = None
+        for el in self.nodes:
+            if isinstance(el, ParallelLayer):
+                res.extend([n for n in el.get_nodes() if n != last_node])
+            else:
+                last_node = el
+                res.append(el)
+
+        return res
 
     def __str__(self):
-        return self.parsed if self.parsed else self.org
+        return f"{' -> '.join([el.name for el in self.nodes])}"
 
 
 @dataclass
-class Layer:
-    edges: Dict[str, Edge]
+class Workflow:
+    name: str
+    layers: OrderedDict[str, Union[Layer, ParallelLayer]]
+    _nodes: Dict[str, Node] = field(default_factory=lambda: {})
+
+    def __str__(self):
+        return f"{' -> '.join([el for el in self.layers])}"
+
+    @property
+    def nodes(self):
+        if not self._nodes:
+            last_layer = self.layers[next(reversed(self.layers))]
+
+            if isinstance(last_layer, ParallelLayer):
+                last_layer.set_nodes_layer(at_node=False, to_node=True)
+
+            for i, (l_name, layer) in enumerate(self.layers.items()):
+                layer.set_nodes_layer()
+                for node in layer.get_nodes():
+                    if node.name not in self._nodes:
+                        self._nodes[node.name] = node
+
+        return self._nodes
+
+    def get_graph(self) -> nx.DiGraph:
+        print(f"Building graph for layers {list(self.layers.keys())}")
+        # set nodes
+        _ = self.nodes
+        graph = nx.DiGraph()
+
+        for i, (l_name, layer) in enumerate(self.layers.items()):
+            if isinstance(layer, ParallelLayer):
+                graph.add_node(l_name, type='layer', layer=l_name)
+                for track in layer.tracks.values():
+                    if graph.has_node(track[-1].at.name):
+                        graph.add_edge(track[-1].at.name, l_name)
+
+                    if i == 0 and track[-1].to.layer != l_name and not graph.has_node(track[-1].to.name):
+                        graph.add_node(track[-1].to.name, type='node', layer=track[-1].to.layer)
+                        graph.add_edge(l_name, track[-1].to.name)
+            else:
+                for e_name, edge in layer.edges.items():
+                    graph.add_node(edge.at.name, type='node', layer=l_name)
+                    graph.add_node(edge.to.name, type='node', layer=l_name)
+                    graph.add_edge(edge.at.name, edge.to.name)
+
+        return graph
+
+    def get_all_paths(self, graph: nx.DiGraph) -> List[DAGPath]:
+        if len(graph.edges) == 0:
+            for n, data in graph.nodes(data=True):
+                if data['type'] == 'layer':
+                    return [DAGPath([self.layers[n]])]
+                else:
+                    return [DAGPath([self.nodes[n]])]
+
+        nodes = {k: v['type'] for k, v in graph.nodes(data=True)}
+        roots = [node for node in graph.nodes if graph.in_degree(node) == 0]
+        leaves = [node for node in graph.nodes if graph.out_degree(node) == 0]
+        paths = []
 
-    def traverse(self, nodes: Dict[str, Plugin]):
-        return [(nodes[edge.node], edge) for edge in self.edges.values()]
+        for root in roots:
+            for leaf in leaves:
+                for path in nx.all_simple_paths(graph, root, leaf):
+                    paths.append(DAGPath([self.nodes[n] if nodes[n] == 'node' else self.layers[n] for n in path]))
+
+        return paths
 
 
 @dataclass
@@ -168,22 +250,41 @@ class Connectors:
     sinks: Dict[str, Dict[str, Connector]] = field(default_factory=lambda: {})
     sources: Dict[str, Dict[str, Connector]] = field(default_factory=lambda: {})
 
+    def init_sinks(self, name: str):
+        if name not in self.sinks:
+            self.sinks[name] = {}
+
+    def init_sources(self, name: str):
+        if name not in self.sources:
+            self.sources[name] = {}
+
+    def init_links(self, at_node: str, to_node: str, links: dict):
+        connector = Connector(source=at_node, sink=to_node, attrs={}, links=links)
+        self.sources[at_node][to_node] = connector
+        self.sources[to_node][at_node] = connector
+
     def __getitem__(self, sink_key: Tuple[str, str]):
         sink, key = sink_key
 
+        # todo: check this code if is used
         for connector in self.sinks[sink].values():
-            if key in connector.links.values():
-                return connector[key]
+            if connector.sink == sink:
+                for k, v in connector.links.items():
+                    if connector.sink == sink and key == v:
+                        return connector[v]
 
         for connector in self.sources[sink].values():
-            if key in connector.links.keys():
-                return connector[key]
+            if connector.sink == sink:
+                for k, v in connector.links.items():
+                    if key == v:
+                        return connector[v]
 
     def __setitem__(self, source_attr: Tuple[str, str], value: Any):
         source, attr = source_attr
 
         for name, connector in self.sources[source].items():
-            connector[attr] = value
+            if attr in connector.links:
+                connector[connector.links[attr]] = value
 
     def has_values(self, source: str) -> bool:
         """
@@ -206,86 +307,10 @@ def has_source(self, node: str, attr: str = None):
                 if attr in connector.attrs:
                     return True
             return False
-        return node in self.sources
-
-
-@dataclass
-class Pipeline:
-    """
-        Data object representing the pipeline
-    """
-    layers: dict
-    nodes: dict
-    workflows: dict
-
-    def unpack(self, workflow: str):
-        return {name: edge for l_name, layer in self.layers.items() if l_name in self.workflows[workflow] for name, edge in layer.edges.items()}
-
-    def match(self, placeholders: list, sink_attrs: list):
-        """
-            Checks whether the placeholders match sink attributes.
-        """
-        for attr in sink_attrs:
-            if attr not in placeholders:
-                return False
-        return True
-
-    def link(self, workflow: str, node_handlers: dict, connectors: Connectors = None) -> Connectors:
-        """
-            creates the respective connectors
-        """
-        if connectors is None:
-            connectors = Connectors()
-            sources = {}
-        else:
-            sources = connectors.sources.copy()
-
-        for _, edge in self.unpack(workflow).items():
-            if edge.sources:
-                if edge.name not in connectors.sources:
-                    connectors.sources[edge.name] = {}
-
-                # If node has attributes, init connector with the value of the attributes
-                if edge.name in sources:
-                    sources[edge.name] = {attr: getattr(node_handlers[edge.name], attr, None) for attr in edge.sources}
-
-            for source, links in edge.sinks.items():
-                if source == edge.name:
-                    raise ValueError(f"Plugin {source} cannot reference itself.")
-
-                if source not in sources:
-                    raise TenetError(f"source {source} must be defined before sink {edge.name}")
-
-                if source in sources:
-                    connector = Connector(source=source, sink=edge.name, attrs=sources[source], links=links)
-                    connectors.sources[source][edge.name] = connector
-
-                    if edge.name not in connectors.sinks:
-                        connectors.sinks[edge.name] = {}
-
-                    connectors.sinks[edge.name][source] = connector
-
-        return connectors
-
-    def walk(self, layers: list):
-        """
-            Walks the edges and returns list with the traversal. Initializes edge connectors.
-        """
-        # TODO: traversal include parallel execution
-        traversal = []
-
-        for el in layers:
-            if el in self.layers:
-                traversal.extend(self.layers[el].traverse(self.nodes))
-            elif el in self.nodes:
-                traversal.append((self.nodes[el], None))
-            else:
-                raise ValueError(f"{el} not found.")
-
-        return traversal
+        return node in self.sources and self.sources[node]
 
 
-def parse_pipeline(yaml: dict) -> Pipeline:
+def parse_pipeline(yaml: dict):
     """
         Parse the yaml file into piepline
 
@@ -293,5 +318,6 @@ def parse_pipeline(yaml: dict) -> Pipeline:
         :return: Pipeline object
     """
 
-    return Schema(And({'nodes': _nodes, 'layers': _layers, 'workflows': _workflows},
-                      Use(lambda pipe: Pipeline(**pipe)))).validate(yaml)
+    return Schema(And({'nodes': _nodes, 'edges': _edges, 'links': _links,
+                       'layers': Schema(And({str: Or(list, [list])})),
+                       'workflows': Schema(And({str: list}))})).validate(yaml)
diff --git a/tenet/handlers/container.py b/tenet/handlers/container.py
index 563f06a..9655028 100644
--- a/tenet/handlers/container.py
+++ b/tenet/handlers/container.py
@@ -82,7 +82,8 @@ def execute_cmd(self, cmd: ContainerCommand) -> bool:
         cmd.skip = True
         return True
 
-    def run_cmds(self, container_id: str, cmds: List[ContainerCommand]) -> Tuple[bool, List[CommandData]]:
+    def run_cmds(self, container_id: str, cmds: List[ContainerCommand], supress_err: bool = False) \
+            -> Tuple[bool, List[CommandData]]:
         """
             Run commands inside the specified container.
         """
@@ -104,7 +105,7 @@ def run_cmds(self, container_id: str, cmds: List[ContainerCommand]) -> Tuple[boo
                 if cmd.parse_fn is not None:
                     cmd_data.parsed_output = cmd.parse_fn(cmd_data.output)
 
-            if cmd_data.error or cmd_data.return_code != 0:
+            if (cmd_data.error or cmd_data.return_code != 0) and (not supress_err):
                 return False, cmds_data
 
         return True, cmds_data
diff --git a/tenet/handlers/github.py b/tenet/handlers/github.py
index 83369aa..3aa16f9 100644
--- a/tenet/handlers/github.py
+++ b/tenet/handlers/github.py
@@ -185,6 +185,10 @@ def has_rate_available(self):
     @property
     def git_api(self):
         with self.lock:
+            if not self._tokens:
+                tokens = self.app.pargs.tokens.split(',')
+                self._tokens = deque(tokens, maxlen=len(tokens))
+
             if not self._git_api:
                 self._git_api = Github(self._tokens[0])
                 self._tokens.rotate(-1)
@@ -204,18 +208,6 @@ def git_api(self):
         with self.lock:
             self._git_api = None
 
-    @property
-    def tokens(self):
-        self._tokens.rotate(-1)
-        return self._tokens[0]
-
-    @tokens.setter
-    def tokens(self, value: Union[str, list]):
-        if isinstance(value, str):
-            value = [value]
-
-        self._tokens = deque(value, maxlen=len(value))
-
     @staticmethod
     def parse_commit_sha(commit_sha: Union[list, str]) -> list:
         commit_hashes = []
@@ -418,10 +410,15 @@ def get_commit_parents(commit: Commit) -> set:
     def get_commit_comments(self, commit: Commit, raise_err: bool = False) -> dict:
         comments, count = {}, 1
         err_msg = None
-        commit_comments = []
 
         try:
-            commit_comments = commit.get_comments()
+            for comment in commit.get_comments():
+                comments[f'com_{count}'] = {
+                    'author': comment.user.login,
+                    'datetime': comment.created_at.strftime("%m/%d/%Y, %H:%M:%S"),
+                    'body': comment.body.strip()
+                }
+                count += 1
         except RateLimitExceededException as rle:
             err_msg = f"Rate limit exhausted: {rle}"
         #except Exception:
@@ -432,14 +429,6 @@ def get_commit_comments(self, commit: Commit, raise_err: bool = False) -> dict:
                 raise TenetError(err_msg)
 
             self.app.log.error(err_msg)
-        # TODO: add check for rate limit
-        for comment in commit_comments:
-            comments[f'com_{count}'] = {
-                'author': comment.user.login,
-                'datetime': comment.created_at.strftime("%m/%d/%Y, %H:%M:%S"),
-                'body': comment.body.strip()
-            }
-            count += 1
 
         return comments
 
diff --git a/tenet/handlers/node.py b/tenet/handlers/node.py
index 9245346..858b100 100644
--- a/tenet/handlers/node.py
+++ b/tenet/handlers/node.py
@@ -4,9 +4,10 @@
 import pandas as pd
 from cement import Handler
 
-from tenet.core.exc import Skip
+from tenet.core.exc import Skip, TenetError
 from tenet.core.interfaces import PluginsInterface
-from tenet.data.schema import Edge
+from tenet.data.plugin import Sources, Sinks
+from tenet.data.schema import Node
 
 
 class NodeHandler(PluginsInterface, Handler):
@@ -17,8 +18,10 @@ def __init__(self, **kw):
         super().__init__(**kw)
         self.path = None
         self.output = None
-        self.edge = None
         self.node = None
+        self.edge = None
+        self.sources = Sources()
+        self.sinks = Sinks()
 
     def load_dataset(self, suffix: str = '.csv', terminate: bool = False):
         """
@@ -33,8 +36,8 @@ def load_dataset(self, suffix: str = '.csv', terminate: bool = False):
             try:
                 return pd.read_csv(str(self.output))
             except pd.errors.EmptyDataError as ede:
-                self.app.log.warning(ede)
-                return self.output.open(mode='r').read()
+                self.app.log.warning(f"Empty Data Error: {ede}")
+                return pd.DataFrame()
 
         self.app.log.error("dataset not found")
 
@@ -43,40 +46,47 @@ def load_dataset(self, suffix: str = '.csv', terminate: bool = False):
 
         return None
 
+    def check_sink(self, name: str):
+        if not self.sinks[name]:
+            raise TenetError(f"Sink '{name}' of '{self.node.name}' node not instantiated")
+
+        # check if path exists
+        if isinstance(self.sinks[name], Path) and not self.sinks[name].exists():
+            raise TenetError(f"Path '{self.sinks[name]}' for sink '{name}' of '{self.node.name}' node not found.")
+
     def get(self, attr: str, default: Any = None):
         try:
-            return self.app.connectors[self.edge.name, attr]
+            self.sinks[attr] = self.app.connectors[self.node.name, attr]
+            self.check_sink(attr)
+            return self.app.connectors[self.node.name, attr]
         except KeyError as ke:
-            self.app.log.warning(f"Key {ke} not found for edge {self.edge.name}")
-            return default
-
-    def get_sinks(self):
-        return self.app.connectors.sinks[self.edge.name]
+            if not default:
+                raise TenetError(f"Sink '{ke}' of '{self.node.name}' node not found")
 
-    def set(self, attr: str, value: Any, skip: bool = True):
-        self.app.connectors[self.edge.name, attr] = value
+            self.sinks[attr] = default
 
-        if self.has_dataset and self.app.connectors.has_source(self.edge.name) and \
-                self.app.connectors.has_values(self.edge.name) and skip:
-            raise Skip(f"Connectors for source \"{self.edge.name}\" are instantiated and exist.")
+    def set(self, attr: str, value: Any):
+        self.app.connectors[self.node.name, attr] = value
+        self.sources[attr] = value
 
-    def load(self, edge: Edge, dataset_name: str, ext: str = '.csv'):
+    def load(self, node: Node, dataset_name: str, ext: str = '.csv'):
         """
             Loads all the related paths to the workflow
 
-            :param edge: edge object
+            :param node: node object
+            :param dataset_name: name of the dataset
+            :param ext: extension of the dataset
         """
 
-        paths = {p.name: p for p in self.app.workdir.iterdir()}
+        self.path = self.app.workdir / node.layer / node.name
 
-        if edge.name in paths:
-            self.app.log.info(f"Loading node {edge.name}")
-            self.path = Path(paths[edge.name])
+        if self.path.exists():
+            self.app.log.info(f"Loading node {node.name}")
+            self.path = Path(self.path)
             # TODO: include correct datasets, and add the layer as well
         else:
-            self.app.log.info(f"Making directory for {edge.name}.")
-            self.path = Path(self.app.workdir / edge.name)
-            self.path.mkdir()
+            self.app.log.info(f"Making directory for {node.name}.")
+            self.path.mkdir(parents=True, exist_ok=True)
 
         self.output = self.path / f"{dataset_name}{ext}"
 
@@ -85,7 +95,6 @@ def has_dataset(self):
         """
             Checks whether output exists.
         """
-
         return self.output.exists()
 
     @property
@@ -93,7 +102,7 @@ def is_skippable(self):
         """
             Checks whether node execution can be skipped, that is, output exists and node does not have dependencies.
         """
-        return self.has_dataset and not self.app.connectors.has_source(self.edge.name)
+        return self.has_dataset and len(self.sources) == 0
 
     def __str__(self):
-        return self.edge.name if self.edge else ""
+        return self.node.name if self.node else ""
diff --git a/tenet/handlers/pipeline.py b/tenet/handlers/pipeline.py
new file mode 100644
index 0000000..1726049
--- /dev/null
+++ b/tenet/handlers/pipeline.py
@@ -0,0 +1,119 @@
+import yaml
+
+from pathlib import Path
+from cement import Handler
+from schema import SchemaError
+from collections import OrderedDict
+from typing import List
+
+from tenet.core.exc import TenetError
+from tenet.core.interfaces import HandlersInterface
+from ..data.schema import parse_pipeline, Connectors, Layer, ParallelLayer, Node, Workflow, Edge
+
+
+class PipelineHandler(HandlersInterface, Handler):
+    """
+        Workflow handler to execute the pipeline.
+    """
+
+    class Meta:
+        label = 'pipeline'
+
+    def __init__(self, **kw):
+        super().__init__(**kw)
+        self.edges = {}
+        self.nodes = {}
+        self.workflows = {}
+
+    def _init_nodes(self, nodes: list):
+        for node in nodes:
+            for k, v in node.items():
+                self.nodes[k] = Node(name=k, **v)
+
+    def _init_edges(self, edges: list):
+        for edge_el in edges:
+            for e_name, edge in edge_el.items():
+                self.edges[e_name] = Edge(name=e_name, at=self.nodes[edge['at']], to=self.nodes[edge['to']])
+
+    def _init_workflows(self, workflows: dict):
+        for w_name, workflow in workflows.items():
+            layers = OrderedDict()
+
+            for layer in workflow:
+                if layer not in self.layers:
+                    raise TenetError(f"Layer {layer} not defined")
+
+                layers[layer] = self.layers[layer]
+
+            self.workflows[w_name] = Workflow(name=w_name, layers=layers)
+
+    def _init_layers(self, layers):
+        self.layers = {}
+
+        for l_name, l_edges in layers.items():
+            if any(isinstance(x, list) for x in l_edges):
+                self.layers[l_name] = ParallelLayer(tracks={i: [self.edges[e] for e in x] for i, x in enumerate(l_edges)},
+                                                    name=l_name)
+            else:
+                self.layers[l_name] = Layer(edges={le: self.edges[le] for le in l_edges}, name=l_name)
+
+    def _init_connectors(self):
+        self.app.extend('connectors', Connectors())
+
+        for name, node in self.nodes.items():
+            self.app.connectors.init_sources(name)
+            self.app.connectors.init_sinks(name)
+
+    def _init_links(self, links):
+        """
+            creates the respective connectors
+        """
+        self.app.log.info("Linking nodes")
+        parsed_links = {}
+
+        for at_node, at_node_links in links.items():
+            for at_node_source, to_node_links in at_node_links.items():
+                for to_node_sink, to_nodes in to_node_links.items():
+                    for to_node in to_nodes:
+                        if at_node not in parsed_links:
+                            parsed_links[at_node] = {}
+                        if to_node not in parsed_links[at_node]:
+                            parsed_links[at_node][to_node] = {}
+
+                        parsed_links[at_node][to_node][at_node_source] = to_node_sink
+
+        for at_node, to_nodes in parsed_links.items():
+            for to_node, at_node_links in to_nodes.items():
+                self.app.connectors.init_links(at_node=at_node, to_node=to_node, links=at_node_links)
+
+    def load(self, path: Path):
+        """
+            Loads and initializes pipeline
+        """
+
+        with path.open(mode="r") as stream:
+            try:
+                self.app.log.info(f"Parsing pipeline file: {path}")
+                pipeline = parse_pipeline(yaml.safe_load(stream))
+            except SchemaError as se:
+                raise TenetError(str(se))
+
+        self._init_nodes(pipeline['nodes'])
+        self._init_edges(pipeline['edges'])
+        self._init_layers(pipeline['layers'])
+        self._init_workflows(pipeline['workflows'])
+
+        self._init_connectors()
+        self._init_links(pipeline['links'])
+
+    def get_nodes(self, nodes: str) -> List[Node]:
+        return [self.nodes[node] for node in nodes]
+
+    def get_nodes_from_edges(self, edges: List[Edge]) -> List[Node]:
+        nodes = []
+        for edge in edges:
+            for n in edge.to_list():
+                if self.nodes[n] not in nodes:
+                    nodes.append(self.nodes[n])
+
+        return nodes
diff --git a/tenet/handlers/plugin.py b/tenet/handlers/plugin.py
index 031c18d..5dd1131 100644
--- a/tenet/handlers/plugin.py
+++ b/tenet/handlers/plugin.py
@@ -123,6 +123,20 @@ def run(self, dataset: pd.DataFrame, **kwargs) -> Union[pd.DataFrame, None]:
         """
         pass
 
+    @abstractmethod
+    def set_sources(self):
+        """
+            Sets sources of the plugin for sinks of other plugins
+        """
+        pass
+
+    @abstractmethod
+    def get_sinks(self):
+        """
+            Inits sinks of the plugin with sources of other plugins
+        """
+        pass
+
     def plot(self, dataset: pd.DataFrame, **kwargs):
         """
             Plots resulting dataframe
diff --git a/tenet/handlers/workflow.py b/tenet/handlers/workflow.py
index 1a9261e..d968fcb 100644
--- a/tenet/handlers/workflow.py
+++ b/tenet/handlers/workflow.py
@@ -1,16 +1,18 @@
-from typing import Tuple
-
 import pandas as pd
 import tqdm
 
+from typing import Tuple, List
+
 from cement import Handler
 from pathlib import Path
 
 from tenet.core.exc import Skip, TenetError
 from tenet.core.interfaces import HandlersInterface
-from tenet.data.schema import Edge
+from tenet.data.schema import Node, DAGPath, ParallelLayer
 from collections import OrderedDict
 
+from tenet.handlers.plugin import PluginHandler
+
 
 class WorkflowHandler(HandlersInterface, Handler):
     """
@@ -22,77 +24,118 @@ class Meta:
 
     def __init__(self, **kw):
         super().__init__(**kw)
-        self.traversal = OrderedDict()
+        self.current_nodes = OrderedDict()
+        self.previous_node = None
 
-    def load(self, workflow_name, dataset_path: Path, layers: list):
+    def _load(self, dataset_path: Path, path: DAGPath):
         """
             Loads and initializes plugins
         """
-        self.app.log.info(f"Loading {layers} layers for {workflow_name} workflow")
-        for node, edge in tqdm.tqdm(self.app.pipeline.walk(layers), desc="Initializing plugins", colour='blue'):
-            if not edge:
-                edge = Edge(name=node.name, node=node.name)
 
-            self.app.log.info(f"Traversing edge {edge.name}")
+        self.app.log.info(f"Loading nodes in {path} path...")
+
+        for node in tqdm.tqdm(path.flatten(), desc="Initializing plugins", colour='blue'):
+            if node.name not in self.current_nodes:
+                node_handler = self.app.get_plugin_handler(node.plugin)
+                node_handler.load(node, dataset_name=dataset_path.stem)
+                node_handler.node = node
+            else:
+                node_handler = self.current_nodes[node.name]
+
+            self.current_nodes[node.name] = node_handler
 
-            node_handler = self.app.get_plugin_handler(node.name)
-            node_handler.load(edge, dataset_name=dataset_path.stem)
+    def _exec_node(self, node_handler: PluginHandler, dataframe: pd.DataFrame) -> Tuple[pd.DataFrame, Path]:
+        self.app.log.info(f"Running node {node_handler}")
+        node_handler.get_sinks()
+        node_handler.set_sources()
 
-            node_handler.node = node
-            node_handler.edge = edge
-            self.traversal[edge.name] = node_handler
+        if node_handler.is_skippable:
+            self.app.log.info(f"{node_handler.node.name}: dataset {node_handler.output} exists.")
+            dataframe = pd.read_csv(node_handler.output)
+            dataset_path = node_handler.output
 
-        self.app.log.info("Linking nodes")
-        # Instantiates the connectors for the nodes
-        if hasattr(self.app, 'connectors'):
-            self.app.connectors = self.app.pipeline.link(workflow_name, self.app.executed_edges, self.app.connectors)
+            return dataframe, dataset_path
+
+        kwargs = node_handler.node.kwargs.copy()
+
+        if node_handler.node.kwargs:
+            kwargs.update(node_handler.node.kwargs)
+
+        if node_handler.has_dataset and node_handler.sources.is_init() and \
+                self.app.connectors.has_values(node_handler.node.name):
+
+            self.app.log.warning(f"Connectors for source \"{node_handler.node.name}\" are instantiated and exist.")
+            self.app.log.warning(f"Skipping {node_handler}.")
+            dataframe = node_handler.load_dataset()
+            dataset_path = node_handler.output
         else:
-            self.app.extend('connectors', self.app.pipeline.link(workflow_name, self.traversal))
+            dataframe = node_handler.run(dataset=dataframe, **kwargs)
+            dataset_path = node_handler.output
+
+            if dataframe is not None:
+                dataframe.to_csv(str(node_handler.output), index=False)
+                self.app.log.info(f"Saving dataset {node_handler.output}.")
+            else:
+                raise TenetError(f"Node {node_handler} returned no dataframe. Stopping execution.")
 
-    def __call__(self, dataset_path: Path) -> Tuple[pd.DataFrame, Path]:
+        self.previous_node = node_handler.node
+
+        return dataframe, dataset_path
+
+    def _exec_layer(self, layer: ParallelLayer, dataframe: pd.DataFrame, dataset_path: Path) \
+            -> Tuple[pd.DataFrame, Path]:
+        # TODO: verify connection between edges in tracks
+        tracks_outcome = []
+        layer_dataset_path = self.app.workdir / layer.name / dataset_path.name
+
+        if not layer_dataset_path.exists():
+            for t_name, edges in layer.tracks.items():
+                self.app.log.info(f"Executing track {t_name}")
+                track_outcome = [(dataframe, dataset_path)]
+
+                for edge in edges:
+                    for node in edge.to_list():
+                        # TODO: change; here we skip the previous executed node that leaded to this layer
+                        if node == self.previous_node:
+                            continue
+                        node_handler = self.current_nodes[node.name]
+                        track_dataframe, track_dataset_path = self._exec_node(node_handler, track_outcome[-1][0])
+                        track_outcome.append((track_dataframe, track_dataset_path))
+
+                        if not self.app.pargs.suppress_plot:
+                            self.app.log.info(f"{node_handler.node.name} plotting...")
+                            node_handler.plot(dataframe)
+
+                tracks_outcome.append(track_outcome[-1])
+
+            dataframe = pd.concat([d for d, dp in tracks_outcome], ignore_index=True)
+            dataframe.to_csv(str(layer_dataset_path), index=False)
+            self.app.log.info(f"{len(dataframe)} entries after merging {list(layer.tracks.keys())} tracks;")
+        else:
+            # TODO: should load the sources from executed nodes
+            self.app.log.warning(f"Dataset for '{layer.name}' layer exists, loading {layer_dataset_path} ...")
+            dataframe = pd.read_csv(str(layer_dataset_path))
+
+        self.previous_node = None
+
+        return dataframe, layer_dataset_path
+
+    def __call__(self, dataset_path: Path, dag_path: DAGPath) -> Tuple[pd.DataFrame, Path]:
+        self._load(dataset_path, dag_path)
         dataframe = pd.read_csv(str(dataset_path), sep='\t' if dataset_path.suffix == '.tsv' else ',')
 
         if dataframe.empty:
             raise TenetError(f"Dataset is empty.")
 
-        while tqdm.tqdm(self.traversal, desc="Executing pipeline", colour='green'):
-            node_name, node_handler = self.traversal.popitem(last=False)
-
-            self.app.log.info(f"Running node {node_handler}")
-
-            if node_handler.is_skippable:
-                self.app.log.info(f"{node_handler.edge.name}: dataset {node_handler.output} exists.")
-                dataframe = pd.read_csv(node_handler.output)
-                dataset_path = node_handler.output
+        for el in tqdm.tqdm(dag_path.nodes, desc="Executing pipeline", colour='green'):
+            if not isinstance(el, Node):
+                dataframe, dataset_path = self._exec_layer(el, dataframe, dataset_path)
+            else:
+                node_handler = self.current_nodes[el.name]
+                dataframe, dataset_path = self._exec_node(node_handler, dataframe)
 
                 if not self.app.pargs.suppress_plot:
-                    self.app.log.info(f"{node_handler.edge.name} plotting...")
+                    self.app.log.info(f"{node_handler.node.name} plotting...")
                     node_handler.plot(dataframe)
-            else:
-                kwargs = node_handler.node.kwargs.copy()
-
-                if node_handler.edge.kwargs:
-                    kwargs.update(node_handler.edge.kwargs)
-
-                try:
-                    dataframe = node_handler.run(dataset=dataframe, **kwargs)
-                    dataset_path = node_handler.output
-
-                    if dataframe is not None:
-                        dataframe.to_csv(str(node_handler.output), index=False)
-                        self.app.log.info(f"Saving dataset {node_handler.output}.")
-                    else:
-                        raise TenetError(f"Node {node_handler} returned no dataframe. Stopping execution.")
-                    if not self.app.pargs.suppress_plot:
-                        node_handler.plot(dataframe)
-                except Skip as se:
-                    self.app.log.warning(f"{se} Skipping {node_handler}.")
-                    dataframe = node_handler.load_dataset()
-                    dataset_path = node_handler.output
-
-                    if not self.app.pargs.suppress_plot:
-                        node_handler.plot(dataframe)
-
-            self.app.executed_edges[node_name] = node_handler
 
         return dataframe, dataset_path
diff --git a/tenet/main.py b/tenet/main.py
index 8635868..7008d00 100644
--- a/tenet/main.py
+++ b/tenet/main.py
@@ -13,6 +13,7 @@
 from tenet.handlers.command import CommandHandler
 from tenet.handlers.runner import MultiTaskHandler
 from tenet.handlers.workflow import WorkflowHandler
+from tenet.handlers.pipeline import PipelineHandler
 from tenet.handlers.plugin_loader import PluginLoader
 from tenet.handlers.container import ContainerHandler
 from tenet.handlers.code_parser import CodeParserHandler
@@ -72,7 +73,7 @@ def get_absolute_path(package, file_path):
         # register handlers
         handlers = [
             Base, Plugin, CWE, PluginLoader, ContainerHandler, CommandHandler, WorkflowHandler, CodeParserHandler,
-            MultiTaskHandler, GithubHandler, CWEListHandler, FileParserHandler, SamplingHandler
+            MultiTaskHandler, GithubHandler, CWEListHandler, FileParserHandler, SamplingHandler, PipelineHandler
         ]
 
     def get_config(self, key: str):
@@ -91,7 +92,10 @@ def get_plugin_handler(self, name: str):
         """
 
         try:
-            self.plugin.load_plugin(name)
+
+            if name not in self.plugin.get_loaded_plugins():
+                self.plugin.load_plugin(name)
+
             plugin = self.handler.resolve('plugins', name)
             plugin.__init__()
             plugin._setup(self)
diff --git a/tenet/plugins/astminer.py b/tenet/plugins/astminer.py
index f9d69df..65d685f 100644
--- a/tenet/plugins/astminer.py
+++ b/tenet/plugins/astminer.py
@@ -12,8 +12,8 @@
 from tenet.data.schema import ContainerCommand
 from tenet.handlers.plugin import PluginHandler
 
-re_plain = '(?P<label>(\w+)),(?P<hash>(\w+)),(?P<pair_hash>(\w*)),(?P<fpath>([\w\/\.\_\-]+)),(?P<sline>(\d+)),(?P<scol>(\d+)),(?P<eline>(\d+)),(?P<ecol>(\d+))'
-re_context_paths = '(?P<fpath>([\w\/\.\_-]+))\_(?P<sline>(\d+))\_(?P<scol>(\d+))\_(?P<eline>(\d+))\_(?P<ecol>(\d+)) (?P<label>(\w+)) (?P<hash>(\w+)) (?P<pair_hash>(\w*)) (?P<context_paths>([\w ,|]+))'
+re_plain = '(?P<label>(\w+)),(?P<hash>(\w+)),(?P<pair_hash>(\w*)),(?P<fpath>([\@\w\/\.\_\-]+)),(?P<sline>(\d+)),(?P<scol>(\d+)),(?P<eline>(\d+)),(?P<ecol>(\d+))'
+re_context_paths = '(?P<fpath>([\@\w\/\.\_-]+))\_(?P<sline>(\d+))\_(?P<scol>(\d+))\_(?P<eline>(\d+))\_(?P<ecol>(\d+)) (?P<label>(\w+)) (?P<hash>(\w+)) (?P<pair_hash>(\w*)) (?P<context_paths>([\w ,|]+))'
 
 
 class ASTMinerHandler(PluginHandler):
@@ -26,27 +26,26 @@ class Meta:
 
     def __init__(self, **kw):
         super().__init__(**kw)
-        self.path_context_file = None
         self.extract_cp = None
         self.raw_files_path = None
 
+    def set_sources(self):
+        self.set('path_contexts', self.path / 'path_contexts.c2s')
+        self.set('dataset', self.output)
+
+    def get_sinks(self):
+        self.get('raw_fn_bounds_file')
+        self.get('raw_files_path')
+
     def run(self, dataset: pd.DataFrame, extract_cp: bool = True, image_name: str = "astminer",
             max_old_space_size: int = 8192, mutations: bool = False, **kwargs) -> Union[pd.DataFrame, None]:
         """
             runs the plugin
         """
         self.extract_cp = extract_cp
-        self.path_context_file = (self.path / 'path_contexts.c2s')
-        self.set('path_contexts', self.path_context_file)
-        self.set('dataset', self.output)
-
-        if not self.get('raw_fn_bounds_file'):
-            self.app.log.warning(f"path to dataset with fn boundaries not instantiated.")
-            return None
-
-        self.raw_files_path = Path(str(self.get('raw_files_path')).replace(str(self.app.workdir), str(self.app.bind)))
+        self.raw_files_path = Path(str(self.sinks['raw_files_path']).replace(str(self.app.workdir), str(self.app.bind)))
 
-        if not self.path_context_file.exists():
+        if not self.sources['path_contexts'].exists():
             container = self.container_handler.run(image_name=image_name)
             self.run_astminer(container, max_old_space_size, mutations)
 
@@ -76,7 +75,7 @@ def run_astminer(self, container, max_old_space_size: int, mutations: bool):
         export_cmd = ContainerCommand(org=f"export NODE_OPTIONS=\"--max-old-space-size={max_old_space_size}\"")
 
         raw_fn_bounds_file = Path(
-            str(self.get('raw_fn_bounds_file')).replace(str(self.app.workdir), str(self.app.bind)))
+            str(self.sinks['raw_fn_bounds_file']).replace(str(self.app.workdir), str(self.app.bind)))
 
         max_old_space_size_gb = round(max_old_space_size / 1024)
         max_mem = f"-Xmx{max_old_space_size_gb}g"
@@ -105,7 +104,7 @@ def parse_path_context_file(self):
         results = []
         self.app.log.info(f"Parsing output file {self.path / 'path_contexts.c2s'}")
 
-        with (self.path / 'path_contexts.c2s').open(mode='r') as output:
+        with self.sources['path_contexts'].open(mode='r') as output:
             reg_exp = re_context_paths if self.extract_cp else re_plain
 
             for i, line in enumerate(output.read().splitlines()):
@@ -203,7 +202,7 @@ def get_functions(self, funcs_df: pd.DataFrame):
         funcs_df.reset_index(inplace=True)
 
         for target_file, rows in tqdm(funcs_df.groupby(['owner', 'project', 'version', 'fpath'])):
-            target_file = self.get('raw_files_path') / '/'.join(target_file)
+            target_file = self.sinks['raw_files_path'] / '/'.join(target_file)
 
             if not target_file.exists():
                 self.app.log.warning(f"{target_file} not found")
diff --git a/tenet/plugins/clean_source.py b/tenet/plugins/clean_source.py
index 02305ed..2791e95 100644
--- a/tenet/plugins/clean_source.py
+++ b/tenet/plugins/clean_source.py
@@ -14,6 +14,12 @@ class CleanSource(PluginHandler):
     class Meta:
         label = "clean_source"
 
+    def set_sources(self):
+        pass
+
+    def get_sinks(self):
+        pass
+
     def run(self, dataset: pd.DataFrame, projects_blacklist: list = None, drop_multi_cwe: bool = False,
             drop_unk_cwe: bool = False, **kwargs) -> Union[pd.DataFrame, None]:
         """Cleans and filters the sources
@@ -23,7 +29,6 @@ def run(self, dataset: pd.DataFrame, projects_blacklist: list = None, drop_multi
             drop_multi_cwe (bool): flag to drop CVE samples with multiple CWE-IDs
             drop_unk_cwe (bool): flag to drop CVE samples with unknown CWE-IDs (e.g., NVD-CWE-Other, NVD-CWE-noinfo)
         """
-
         dataset['message'] = dataset['message'].apply(lambda x: self.msg(x))
         no_merge = ~dataset['message'].str.contains("merge message")
         cwe = dataset['cwe_id'].notnull()
diff --git a/tenet/plugins/code2vec.py b/tenet/plugins/code2vec.py
index a86d9c1..7bc2efb 100644
--- a/tenet/plugins/code2vec.py
+++ b/tenet/plugins/code2vec.py
@@ -36,57 +36,28 @@ class Meta:
     def __init__(self, **kw):
         super().__init__(**kw)
 
+    def set_sources(self):
+        model_dir = Path(self.app.workdir, Path(self.path).name)
+        self.set('save_path', f"{model_dir}/saved_model")
+
+    def get_sinks(self):
+        self.get('train_data_path')
+        self.get('val_data_path')
+        self.get('test_data_path')
+
     def run(self, dataset: pd.DataFrame, max_contexts: int = 200, emb_size: int = 128, training: bool = True,
             evaluate: bool = True, image_name: str = "code2vec", **kwargs) -> Union[pd.DataFrame, None]:
         """
             runs the plugin
         """
+        count_labels(self.sinks['train_data_path'], 'train')
+        count_labels(self.sinks['val_data_path'], 'validation')
+        count_labels(self.sinks['test_data_path'], 'test')
 
-        model_dir = Path(self.app.workdir, Path(self.path).name)
-
-        if self.get('save_path'):
-            save_path = self.get('save_path')
-        else:
-            save_path = f"{model_dir}/saved_model"
-
-        self.set('save_path', save_path)
-
-        train_data_path = self.get('train_data_path')
-        val_data_path = self.get('val_data_path')
-        test_data_path = self.get('test_data_path')
-
-        if not train_data_path:
-            self.app.log.warning(f"Train data file not instantiated.")
-            return None
-
-        if not Path(train_data_path).exists():
-            self.app.log.warning(f"Train data file not found.")
-            return None
-
-        if not val_data_path:
-            self.app.log.warning(f"Validation data file not instantiated.")
-            return None
-
-        if not Path(val_data_path).exists():
-            self.app.log.warning(f"Validation data file not found.")
-            return None
-
-        if not test_data_path:
-            self.app.log.warning(f"Test data file not instantiated.")
-            return None
-
-        if not Path(test_data_path).exists():
-            self.app.log.warning(f"Test data file not found.")
-            return None
-
-        count_labels(Path(train_data_path), 'train')
-        count_labels(Path(val_data_path), 'validation')
-        count_labels(Path(test_data_path), 'test')
-
-        val_data_path = val_data_path.replace(str(self.app.workdir), str(self.app.bind))
+        val_data_path = self.sinks['val_data_path'].replace(str(self.app.workdir), str(self.app.bind))
         container = self.container_handler.run(image_name=image_name)
         # TODO: find better way of performing this bind
-        save_path = save_path.replace(str(self.app.workdir), str(self.app.bind))
+        save_path = self.sources['save_path'].replace(str(self.app.workdir), str(self.app.bind))
         dataset_name = Path(val_data_path).stem.split('.')[0]
         data_dir = Path(Path(val_data_path).parent, dataset_name)
         # step = 'train' if train else 'test'
@@ -102,7 +73,7 @@ def run(self, dataset: pd.DataFrame, max_contexts: int = 200, emb_size: int = 12
             cmds.append(ContainerCommand(org=f"{default} --load {save_path} --test {data_dir}.test.c2v",
                                          parse_fn=parse_results, tag='test'))
 
-        outcome, cmd_data = self.container_handler.run_cmds(container.id, cmds)
+        outcome, cmd_data = self.container_handler.run_cmds(container.id, cmds, supress_err=True)
 
         for cd in cmd_data:
             df = pd.DataFrame(cd.parsed_output)
diff --git a/tenet/plugins/code_mining.py b/tenet/plugins/code_mining.py
index 91b5cb7..eb8b311 100644
--- a/tenet/plugins/code_mining.py
+++ b/tenet/plugins/code_mining.py
@@ -23,6 +23,19 @@ def __init__(self, **kw):
         self.max_features: int = 1000
         self.vectorizer_type: str = 'tfidf'
 
+    def set_sources(self):
+        if 'vectorizer_type' in self.node.kwargs:
+            self.vectorizer_type = self.node.kwargs['vectorizer_type']
+
+        self.set('train_sparse_matrix_path', self.path / f"{self.vectorizer_type}_{self.output.stem}_train_features_sparse.npz")
+        self.set('test_sparse_matrix_path', self.path / f"{self.vectorizer_type}_{self.output.stem}_test_features_sparse.npz")
+        self.set('train_labels_path', self.path / f"{self.output.stem}_train_labels.csv")
+        self.set('test_labels_path', self.path / f"{self.output.stem}_test_labels.csv")
+
+    def get_sinks(self):
+        self.get('train_data_path')
+        self.get('test_data_path')
+
     def run(self, dataset: pd.DataFrame, max_features: int = 1000, vectorizer_type: str = 'tfidf',
             drop_ratio: float = None, drop_tag: str = None, remove_comments: bool = False, **kwargs) \
             -> Union[pd.DataFrame, None]:
@@ -32,38 +45,14 @@ def run(self, dataset: pd.DataFrame, max_features: int = 1000, vectorizer_type:
         self.max_features = max_features
         self.vectorizer_type = vectorizer_type
 
-        dataset_name = self.output.stem
-        train_vectorizer_model_path = Path(self.path, f"bow_{vectorizer_type}_{dataset_name}_train.model")
-        test_vectorizer_model_path = Path(self.path, f"bow_{vectorizer_type}_{dataset_name}_test.model")
-        train_vocabulary_path = Path(self.path, f"bow_{vectorizer_type}_{dataset_name}_train_vocab.txt")
-        test_vocabulary_path = Path(self.path, f"bow_{vectorizer_type}_{dataset_name}_test_vocab.txt")
-        train_sparse_matrix_path = Path(self.path, f"{vectorizer_type}_{dataset_name}_train_features_sparse.npz")
-        test_sparse_matrix_path = Path(self.path, f"{vectorizer_type}_{dataset_name}_test_features_sparse.npz")
-        train_labels_path = Path(self.path, f"{dataset_name}_train_labels.csv")
-        test_labels_path = Path(self.path, f"{dataset_name}_test_labels.csv")
-
-        # TODO: fix this
-        #self.set('train_tokenizer_model_path', train_vectorizer_model_path)
-        #self.set('test_tokenizer_model_path', test_vectorizer_model_path)
-        self.set('train_sparse_matrix_path', train_sparse_matrix_path)
-        self.set('test_sparse_matrix_path', test_sparse_matrix_path)
-        self.set('train_labels_path', train_labels_path)
-        self.set('test_labels_path', test_labels_path)
-
-        train_data_path = self.get('train_data_path')
-        test_data_path = self.get('test_data_path')
-
-        if not train_data_path:
-            self.app.log.error(f"Train data path not instantiated")
-            return None
-
-        if not test_data_path:
-            self.app.log.error(f"Test data path not instantiated")
-            return None
+        train_vectorizer_model_path = Path(self.path, f"bow_{vectorizer_type}_{self.output.stem}_train.model")
+        test_vectorizer_model_path = Path(self.path, f"bow_{vectorizer_type}_{self.output.stem}_test.model")
+        train_vocabulary_path = Path(self.path, f"bow_{vectorizer_type}_{self.output.stem}_train_vocab.txt")
+        test_vocabulary_path = Path(self.path, f"bow_{vectorizer_type}_{self.output.stem}_test_vocab.txt")
 
         # get test/train data
-        train_data = pd.read_csv(str(train_data_path))
-        test_data = pd.read_csv(str(test_data_path))
+        train_data = pd.read_csv(str(self.sinks['train_data_path']))
+        test_data = pd.read_csv(str(self.sinks['test_data_path']))
 
         if drop_ratio and 'tag' in train_data and drop_tag in train_data['tag'].unique():
             if drop_ratio < 1:
@@ -92,8 +81,8 @@ def run(self, dataset: pd.DataFrame, max_features: int = 1000, vectorizer_type:
         x_test = test_data.input.to_numpy()
 
         # save labels
-        train_data.label.to_csv(str(train_labels_path))
-        test_data.label.to_csv(str(test_labels_path))
+        train_data.label.to_csv(str(self.sources['train_labels_path']))
+        test_data.label.to_csv(str(self.sources['test_labels_path']))
 
         #if model:
         #    features = apply_model(vectorizer, sentences, model_path=model)
@@ -120,8 +109,8 @@ def run(self, dataset: pd.DataFrame, max_features: int = 1000, vectorizer_type:
         self.app.log.info(f"Train features shape: {train_features.shape}")
         self.app.log.info(f"Test features shape: {test_features.shape}")
 
-        save_npz(str(train_sparse_matrix_path), train_features)
-        save_npz(str(test_sparse_matrix_path), test_features)
+        save_npz(str(self.sources['train_sparse_matrix_path']), train_features)
+        save_npz(str(self.sources['test_sparse_matrix_path']), test_features)
 
         return dataset
 
diff --git a/tenet/plugins/codebert.py b/tenet/plugins/codebert.py
index 1ad46a2..53c09a5 100644
--- a/tenet/plugins/codebert.py
+++ b/tenet/plugins/codebert.py
@@ -1,5 +1,3 @@
-import re
-
 import pandas as pd
 
 from typing import Union, Tuple
@@ -16,10 +14,7 @@ def parse_results(output: str):
     # TODO: Parse output into results
     for line in output.splitlines():
         print(line)
-    #    match = re.search(reg_exp, line)
 
-    #    if match:
-    #        results.append(match.groupdict())
     return results
 
 
@@ -34,75 +29,30 @@ class Meta:
     def __init__(self, **kw):
         super().__init__(**kw)
 
+    def set_sources(self):
+        pass
+
+    def get_sinks(self):
+        self.get('train_lines_path')
+        self.get('val_lines_path')
+        self.get('test_lines_path')
+        self.get('train_offset_file')
+        self.get('val_offset_file')
+        self.get('test_offset_file')
+
     def run(self, dataset: pd.DataFrame, train: bool = True, evaluate: bool = True, gpus: int = 0, max_epochs: int = 20,
             image_name: str = 'codebert', **kwargs) -> Union[pd.DataFrame, None]:
         """
             runs the plugin
         """
 
-        train_lines_path = self.get('train_lines_path')
-        val_lines_path = self.get('val_lines_path')
-        test_lines_path = self.get('test_lines_path')
-
-        train_offset_file = self.get('train_offset_file')
-        val_offset_file = self.get('val_offset_file')
-        test_offset_file = self.get('test_offset_file')
-
-        if not train_lines_path:
-            self.app.log.warning(f"Train lines file not instantiated.")
-            return None
-
-        if not Path(train_lines_path).exists():
-            self.app.log.warning(f"Train lines file not found.")
-            return None
-
-        if not val_lines_path:
-            self.app.log.warning(f"Validation lines file not instantiated.")
-            return None
-
-        if not Path(val_lines_path).exists():
-            self.app.log.warning(f"Validation lines file not found.")
-            return None
-
-        if not test_lines_path:
-            self.app.log.warning(f"Test lines file not instantiated.")
-            return None
-
-        if not Path(test_lines_path).exists():
-            self.app.log.warning(f"Test lines file not found.")
-            return None
-
-        if not train_offset_file:
-            self.app.log.warning(f"Train offset file not instantiated.")
-            return None
-
-        if not Path(train_offset_file).exists():
-            self.app.log.warning(f"Train offset file not found.")
-            return None
-
-        if not val_offset_file:
-            self.app.log.warning(f"Validation offset file not instantiated.")
-            return None
-
-        if not Path(val_offset_file).exists():
-            self.app.log.warning(f"Validation offset file not found.")
-            return None
-
-        if not test_offset_file:
-            self.app.log.warning(f"Test offset file not instantiated.")
-            return None
-
-        if not Path(test_offset_file).exists():
-            self.app.log.warning(f"Test offset file not found.")
-            return None
-
-        train_lines_path = str(train_lines_path).replace(str(self.app.workdir), str(self.app.bind))
-        val_lines_path = str(val_lines_path).replace(str(self.app.workdir), str(self.app.bind))
-        test_lines_path = str(test_lines_path).replace(str(self.app.workdir), str(self.app.bind))
+        train_lines_path = str(self.sinks['train_lines_path']).replace(str(self.app.workdir), str(self.app.bind))
+        val_lines_path = str(self.sinks['val_lines_path']).replace(str(self.app.workdir), str(self.app.bind))
+        test_lines_path = str(self.sinks['test_lines_path']).replace(str(self.app.workdir), str(self.app.bind))
 
-        train_offset_file = str(train_offset_file).replace(str(self.app.workdir), str(self.app.bind))
-        val_offset_file = str(val_offset_file).replace(str(self.app.workdir), str(self.app.bind))
-        test_offset_file = str(test_offset_file).replace(str(self.app.workdir), str(self.app.bind))
+        train_offset_file = str(self.sinks['train_offset_file']).replace(str(self.app.workdir), str(self.app.bind))
+        val_offset_file = str(self.sinks['val_offset_file']).replace(str(self.app.workdir), str(self.app.bind))
+        test_offset_file = str(self.sinks['test_offset_file']).replace(str(self.app.workdir), str(self.app.bind))
 
         container = self.container_handler.run(image_name=image_name)
 
diff --git a/tenet/plugins/codeql.py b/tenet/plugins/codeql.py
index aa2f292..3d19ce7 100644
--- a/tenet/plugins/codeql.py
+++ b/tenet/plugins/codeql.py
@@ -1,9 +1,11 @@
+import ast
+
 import pandas as pd
 import jq
 import json
 import tqdm
 
-from typing import Union
+from typing import Union, Any
 from pathlib import Path
 
 from tenet.core.exc import TenetError
@@ -43,6 +45,15 @@ def __init__(self, **kw):
                      643, 730, 754, 770, 776, 798, 807, 829, 830, 834, 843, 862, 912, 915, 916, 918]
         self.parent_files = True
         self.fix_files = True
+        self.drop_unavailable = False
+        self.drop_cwes = []
+
+    def set_sources(self):
+        self.set('report_file_path', self.path / f"{self.output.stem}_report.json")
+        self.set('codeql_db_path', self.container_handler.working_dir / 'db')
+
+    def get_sinks(self):
+        self.get('raw_files_path')
 
     def parse_language(self, language: str) -> str:
         if language.lower() in self.language_mapping.values():
@@ -53,20 +64,30 @@ def parse_language(self, language: str) -> str:
 
         raise TenetError(f"Programming language {language} not available.")
 
-    def parse_cwe(self, cwe: str):
+    def parse_cwe(self, cwe: str) -> Union[str, None]:
         if 'CWE-' in cwe.upper():
             cwe_number = int(cwe.upper().split('CWE-')[-1])
         else:
             cwe_number = int(cwe)
 
         if cwe_number not in self.cwes:
-            raise TenetError(f"CWE-{cwe_number} not available.")
+            if not self.drop_unavailable:
+                raise TenetError(f"CWE-{cwe_number} not available.")
+            return None
 
         return f'CWE-{cwe_number:03}'
 
+    def parse_target_cwes(self, cwes: Any):
+        cwes = ast.literal_eval(cwes)
+
+        if not isinstance(cwes, str):
+            return [self.parse_cwe(cwe) for cwe in cwes]
+
+        return [self.parse_cwe(cwes)]
+
     def run(self, dataset: pd.DataFrame, image_name: str = 'codeql', language: str = 'javascript',
-            target_cwes: list = None, parent_files: bool = True, fix_files: bool = True,
-            **kwargs) -> Union[pd.DataFrame, None]:
+            target_cwes: list = None, parent_files: bool = True, fix_files: bool = True, drop_unavailable: bool = False,
+            drop_cwes: list = None, **kwargs) -> Union[pd.DataFrame, None]:
         """
             run CodeQL and extracts the labels from its report
             :param dataset: dataset with diff blocks
@@ -75,30 +96,31 @@ def run(self, dataset: pd.DataFrame, image_name: str = 'codeql', language: str =
             :param target_cwes: list of CWEs to analyze for
             :param parent_files: flag to include warnings for parent files
             :param fix_files: flag to include warnings for fix files
+            :param drop_unavailable: flag to drop target CWEs not covered by CodeQL
+            :param drop_cwes: list of cwes to drop, in case codeql raises some error for them
         """
-        codeql_db_path = self.container_handler.working_dir / 'db'
-        report_file = self.path / f"{self.output.stem}_report.json"
         report_file_container = self.container_handler.working_dir / f"{self.output.stem}_report.json"
-        self.set('report_file_path', report_file)
-        self.set('codeql_db_path', codeql_db_path)
 
-        if not self.get('raw_files_path'):
-            self.app.log.warning(f"raw files path not instantiated.")
-            return None
+        self.drop_unavailable = drop_unavailable
+        if drop_cwes:
+            for cwe in drop_cwes:
+                if cwe in self.cwes:
+                    self.cwes.remove(cwe)
+
 
-        raw_files_path = Path(str(self.get('raw_files_path')).replace(str(self.app.workdir), str(self.app.bind)))
+        raw_files_path = Path(str(self.sinks['raw_files_path']).replace(str(self.app.workdir), str(self.app.bind)))
 
         if not target_cwes:
             target_cwes = dataset.label.unique().tolist()
 
-        if not report_file.exists():
+        if not self.sources['report_file_path'].exists():
             # TODO: fix the node name
             container = self.container_handler.run(image_name=image_name)
             language = self.parse_language(language)
-            cwes = [self.parse_cwe(cwe) for cwe in target_cwes]
-            create_cmd = ContainerCommand(org=f"codeql database create {codeql_db_path}")
+            cwes = [parsed_cwe for cwe in target_cwes for parsed_cwe in self.parse_target_cwes(cwe) if parsed_cwe is not None]
+            create_cmd = ContainerCommand(org=f"codeql database create {self.sources['codeql_db_path']}")
             create_cmd.org += f" --threads={self.app.threads} --language={language} --source-root={raw_files_path} 2>&1"
-            analyze_cmd = ContainerCommand(org=f"codeql database analyze {codeql_db_path} --format=sarif-latest")
+            analyze_cmd = ContainerCommand(org=f"codeql database analyze {self.sources['codeql_db_path']} --format=sarif-latest")
             analyze_cmd.org += f" --threads={self.app.threads} --output={report_file_container}"
 
             for cwe in cwes:
@@ -108,11 +130,11 @@ def run(self, dataset: pd.DataFrame, image_name: str = 'codeql', language: str =
             self.container_handler.run_cmds(container.id, [create_cmd, analyze_cmd])
             self.container_handler.stop(container)
 
-        if not report_file.exists():
+        if not self.sources['report_file_path'].exists():
             self.app.log.warning(f"CodeQL report file not found.")
             return None
 
-        with report_file.open(mode='r') as rf:
+        with self.sources['report_file_path'].open(mode='r') as rf:
             json_report = json.loads(rf.read())
             result = jq.all(r'.runs[0].results', json_report)
             data_points = jq.iter(r'.[]', result[0])
diff --git a/tenet/plugins/collector.py b/tenet/plugins/collector.py
index b7c832f..f24abd2 100644
--- a/tenet/plugins/collector.py
+++ b/tenet/plugins/collector.py
@@ -23,21 +23,15 @@ class Meta:
 
     def __init__(self, **kw):
         super().__init__(**kw)
-        self.files_dir: Path = None
         self.diff_dir: Path = None
 
-    def set_dirs(self):
-        # TODO: refactor this
-        self.files_dir = Path(self.path, 'files')
-        check_or_create_dir(self.files_dir)
-        self.set('files_path', self.files_dir)
+    def set_sources(self):
+        self.set('files_path', Path(self.path, 'files'))
 
-        # Save diff texts as files in the directory
-        self.diff_dir = Path(self.path, 'diffs')
-        check_or_create_dir(self.diff_dir)
+    def get_sinks(self):
+        pass
 
-    def run(self, dataset: pd.DataFrame, tokens: list = None, max_size: int = None, **kwargs) \
-            -> Union[pd.DataFrame, None]:
+    def run(self, dataset: pd.DataFrame, max_size: int = None, **kwargs) -> Union[pd.DataFrame, None]:
         """
             runs the plugin
         """
@@ -46,9 +40,11 @@ def run(self, dataset: pd.DataFrame, tokens: list = None, max_size: int = None,
         if not all([col in dataset.columns for col in cols]):
             raise TenetError(f"Missing columns. (Dataset must contain {cols})")
 
-        self.github_handler.tokens = tokens
-        self.set_dirs()
-        self.set('dataset', self.output)
+        check_or_create_dir(self.sources['files_path'])
+
+        # Save diff texts as files in the directory
+        self.diff_dir = Path(self.path, 'diffs')
+        check_or_create_dir(self.diff_dir)
 
         # Select columns: repo, sha_list, parents and ensure no duplicates
         dataset = dataset.drop_duplicates().reset_index(drop=True)
@@ -102,11 +98,11 @@ def parse_diffs(self, owner: str, project: str, fix_sha: str, vuln_sha: str, lab
                 entry = Entry(owner=owner, project=project, a_version=vuln_sha, b_version=fix_sha, label=label,
                               diff_block=diff_block)
                 # Get the contents of the two files using GitHub API
-                a_output_file = self.files_dir / entry.full_a_path
+                a_output_file = self.sources['files_path'] / entry.full_a_path
                 _, entry.a_file_size = self.github_handler.get_file_from_commit(commit=vuln_commit,
                                                                                 repo_file_path=diff_block.a_path,
                                                                                 output_path=a_output_file)
-                b_output_file = self.files_dir / entry.full_b_path
+                b_output_file = self.sources['files_path'] / entry.full_b_path
                 _, entry.b_file_size = self.github_handler.get_file_from_commit(commit=fix_commit,
                                                                                 repo_file_path=diff_block.b_path,
                                                                                 output_path=b_output_file)
diff --git a/tenet/plugins/download.py b/tenet/plugins/download.py
index eaab49c..811cf8c 100644
--- a/tenet/plugins/download.py
+++ b/tenet/plugins/download.py
@@ -19,18 +19,19 @@ class Meta:
 
     def __init__(self, **kw):
         super().__init__(**kw)
-        self.files_dir: Path = None
 
-    def run(self, dataset: pd.DataFrame, tokens: list = None) -> Union[pd.DataFrame, None]:
+    def set_sources(self):
+        self.set('dataset', self.output)
+        self.set('files_path', Path(self.path, 'files'))
+
+    def get_sinks(self):
+        pass
+
+    def run(self, dataset: pd.DataFrame, **kwargs) -> Union[pd.DataFrame, None]:
         """
             runs the plugin
         """
-        self.files_dir = Path(self.path, 'files')
-        check_or_create_dir(self.files_dir)
-        self.set('files_path', self.files_dir)
-        self.github_handler.tokens = tokens
-        self.set('dataset', self.output)
-
+        check_or_create_dir(self.sources['files_path'])
         self.app.log.info(f"Creating {len(dataset)} tasks.")
 
         for row in tqdm(dataset.to_dict(orient='records')):
@@ -45,7 +46,7 @@ def run(self, dataset: pd.DataFrame, tokens: list = None) -> Union[pd.DataFrame,
         return None
 
     def download(self, row: dict):
-        project_path = self.files_dir / row['project_name']
+        project_path = self.sources['files_path'] / row['project_name']
         if 'raw_url_vuln' in row and not pd.isnull(row['raw_url_vuln']):
             vuln_file = LocalGitFile(url=row['raw_url_vuln'], short=Path(row['file_path']), tag='vuln',
                                      path=project_path / row['vuln_commit_hash'] / row['file_path'])
diff --git a/tenet/plugins/file_collector.py b/tenet/plugins/file_collector.py
deleted file mode 100644
index 083adaa..0000000
--- a/tenet/plugins/file_collector.py
+++ /dev/null
@@ -1,95 +0,0 @@
-import time
-
-import pandas as pd
-import requests
-
-from pathlib import Path
-from typing import Union
-
-from tenet.handlers.plugin import PluginHandler
-from tenet.core.diff_labeller.misc import check_or_create_dir, safe_write
-from tenet.data.runner import Task, Runner
-from tenet.handlers.runner import ThreadPoolWorker
-
-
-class FileCollector(PluginHandler):
-    """
-        Github collector plugin
-    """
-
-    class Meta:
-        label = "file_collector"
-
-    def __init__(self, **kw):
-        super().__init__(**kw)
-
-    def run(self, dataset: pd.DataFrame, **kwargs) -> Union[pd.DataFrame, None]:
-        """
-            runs the plugin
-        """
-        files_dir = Path(self.path, 'files')
-        self.set('files_path', files_dir)
-        self.set('dataset', self.output)
-
-        # Save diff texts as files in the directory
-        check_or_create_dir(files_dir)
-
-        runner_data = Runner()
-        threads = self.app.get_config('local_threads')
-
-        # Select columns: repo, sha_list, parents
-        # Ensure no duplicates
-        tasks = []
-        total = len(dataset)
-        # Drop unnamed columns
-        dataset.drop(columns=[c for c in dataset.columns if 'Unnamed:' in c], inplace=True)
-
-        for i, row in dataset.iterrows():
-            self.app.log.info(f"Creating task for repo {row['project_name'] + row['file_path']}. {i}/{total}")
-            task = Task()
-            task['id'] = i
-            task['row'] = row
-            task['files_dir'] = files_dir
-            tasks.append(task)
-
-        worker = ThreadPoolWorker(runner_data, tasks=tasks, threads=threads, logger=self.app.log,
-                                  func=self.parse_diffs_task)
-        worker.start()
-
-        while worker.is_alive():
-            time.sleep(1)
-
-        rows = [res['result'] for res in runner_data.finished if 'result' in res and res['result']]
-
-        if rows:
-            return pd.DataFrame(rows, columns=dataset.columns)
-
-        return None
-
-    def parse_diffs_task(self, task: Task):
-        return self.parse_diffs(row=task['row'], files_dir=task['files_dir'])
-
-    def parse_diffs(self, row: pd.Series, files_dir: Path):
-        file_path = files_dir / row.project_name / row.commit_hash / row.file_path
-        commit_url = f"{row.project}/raw/{row.commit_hash}/{row.file_path}"
-
-        # Get the diff string using GitHub API
-        self.app.log.info(f"Requesting {commit_url}")
-
-        if not file_path.exists():
-            file_path.parent.mkdir(exist_ok=True, parents=True)
-            fstr = requests.get(commit_url).text
-
-            if fstr:
-                # Save the diff string as a file
-
-                with open(file_path, "w") as tmp_file:
-                    self.app.log.info(f"Writing file to {file_path}")
-                    safe_write(tmp_file, fstr, f"{row.project_name}_{row.commit_hash}")
-
-        row.file_path = f"{row.project_name}/{row.commit_hash}/{row.file_path}"
-        return row.to_list()
-
-
-def load(app):
-    app.handler.register(FileCollector)
diff --git a/tenet/plugins/find_references.py b/tenet/plugins/find_references.py
deleted file mode 100644
index 6c9a943..0000000
--- a/tenet/plugins/find_references.py
+++ /dev/null
@@ -1,99 +0,0 @@
-import ast
-import re
-import pandas as pd
-
-from github import Github, GithubException
-from github.GithubException import UnknownObjectException
-from typing import Union
-
-from tenet.handlers.plugin import PluginHandler
-
-
-class FindReferencesHandler(PluginHandler):
-    """
-        Separate plugin
-    """
-
-    def __init__(self, **kw):
-        super().__init__(**kw)
-        self.token = None
-
-    class Meta:
-        label = "find_references"
-
-    def run(self, dataset: pd.DataFrame, token: str = None, column: str = None, **kwargs) -> Union[pd.DataFrame, None]:
-        """
-            runs the plugin
-        """
-        self.token = token
-        self.set('dataset', self.output)
-
-        if not column:
-            self.app.log.warning("Must specify 'column' name")
-            return None
-
-        if column not in dataset.columns:
-            self.app.log.warning(f"Column {column} not found in DataFrame")
-            return None
-
-        filtered_dataset = []
-        git_api = Github(self.token)
-        git_repo_regex = '(?P<host>(git@|https://)([\w\.@]+)(/|:))(?P<owner>[\w,\-,\_]+)/(?P<repo>[\w,\-,\_]+)(.git){0,1}((/){0,1})'
-        fix_hashes = []
-        parent_hashes = []
-        commits = []
-        repos = []
-        projects = []
-
-        for i, el in dataset.iterrows():
-            refs = ast.literal_eval(el[column])
-            commit_fix_hashes = []
-            commit_parent_hashes = []
-            commit_repos = ''
-            commit_project = ''
-
-            for ref in refs:
-                match = re.match(pattern='https://github\.com(?:/[^/]+)*/commit/([0-9a-f]{40})', string=ref)
-
-                if match:
-                    commit_fix_sha = match.group(1)
-                    match = re.match(pattern=git_repo_regex, string=ref)
-                    repo_name = match['repo']
-                    repo_owner = match['owner']
-                    repo_path = f"{repo_owner}/{repo_name}"
-                    self.app.log.info(f"Parsing {commit_fix_sha}")
-
-                    try:
-                        repo = git_api.get_repo(repo_path)
-                        commit_fix_sha = repo.get_commit(sha=commit_fix_sha)
-                    except (UnknownObjectException, GithubException):
-                        self.app.log.error(f"Commit {commit_fix_sha} for repo {repo_path} not found")
-                        continue
-
-                    commit_project = repo_name
-                    commit_repos = repo_path
-                    commit_fix_hashes.append(commit_fix_sha.sha)
-                    commit_parent_hashes.append(commit_fix_sha.parents[0].sha)
-
-            if len(commit_fix_hashes) > 0:
-                filtered_dataset.append(el)
-                repos.append(commit_repos)
-                projects.append(commit_project)
-                fix_hashes.append(str(commit_fix_hashes))
-                parent_hashes.append(str(commit_parent_hashes))
-                commits.append(len(commit_fix_hashes))
-
-        new_dataset = pd.DataFrame(filtered_dataset, columns=list(dataset.columns.values))
-        new_dataset.rename(columns={'id': 'cve_id'})
-        new_dataset.drop(columns=[column])
-        new_dataset['project'] = projects
-        new_dataset['repo'] = repos
-        new_dataset['fix_sha'] = fix_hashes
-        new_dataset['parent_sha'] = parent_hashes
-        new_dataset['commits'] = commits
-
-        return new_dataset
-
-
-def load(app):
-    app.handler.register(FindReferencesHandler)
diff --git a/tenet/plugins/fusion.py b/tenet/plugins/fusion.py
index 4de6c3a..665c66b 100644
--- a/tenet/plugins/fusion.py
+++ b/tenet/plugins/fusion.py
@@ -15,6 +15,13 @@ class Meta:
     def __init__(self, **kw):
         super().__init__(**kw)
 
+    def set_sources(self):
+        pass
+
+    def get_sinks(self):
+        self.get('diff_labelled_data_path')
+        self.get('static_labelled_data_path')
+
     def run(self, dataset: pd.DataFrame, sa_labels: bool = False, da_labels: bool = False,
             **kwargs) -> Union[pd.DataFrame, None]:
         """
@@ -23,19 +30,8 @@ def run(self, dataset: pd.DataFrame, sa_labels: bool = False, da_labels: bool =
             :param sa_labels: flag to return datasets with only static labels
         """
 
-        diff_labelled_data_path = self.get('diff_labelled_data_path')
-        static_labelled_data_path = self.get('static_labelled_data_path')
-
-        if not diff_labelled_data_path:
-            self.app.log.error(f"Diff labelled data path not instantiated")
-            return None
-
-        if not static_labelled_data_path:
-            self.app.log.error(f"Static labelled data path not instantiated")
-            return None
-
-        diff_labelled_data = pd.read_csv(str(diff_labelled_data_path))
-        static_labelled_data = pd.read_csv(str(static_labelled_data_path))
+        diff_labelled_data = pd.read_csv(str(self.sinks['diff_labelled_data_path']))
+        static_labelled_data = pd.read_csv(str(self.sinks['static_labelled_data_path']))
         self.set_labels(diff_labelled_data, label_type='diff')
         self.set_labels(static_labelled_data, label_type='sa')
 
diff --git a/tenet/plugins/generate.py b/tenet/plugins/generate.py
index 7557c9c..a553c59 100644
--- a/tenet/plugins/generate.py
+++ b/tenet/plugins/generate.py
@@ -28,15 +28,20 @@ def __init__(self, **kw):
         self.early_stopping_limit = None
         self.skip_search = None
 
-    def run(self, dataset: pd.DataFrame, scenario: str = 'fix', augment: int = None, tokens: list = None,
-            target_files_factor: int = 3, languages: list = None,  max_commits: int = None, skip_search: bool = False,
+    def set_sources(self):
+        pass
+
+    def get_sinks(self):
+        pass
+
+    def run(self, dataset: pd.DataFrame, scenario: str = 'fix', augment: int = None, target_files_factor: int = 3,
+            languages: list = None,  max_commits: int = None, skip_search: bool = False,
             **kwargs) -> Union[pd.DataFrame, None]:
         """Generates sampling strategy
         Args:
             scenario (str): one of the following scenarios fix, random, controlled
             languages (list): list of target programming languages
             augment (int): factor for augmenting the negative classes in the dataset
-            tokens (list): GitHub API tokens
             skip_search (bool): flag to skip the search for the negative samples
             max_commits (int): limit of commits to search
             target_files_factor (int): the factor used to gather target files for the negative dataset,
@@ -61,8 +66,6 @@ def run(self, dataset: pd.DataFrame, scenario: str = 'fix', augment: int = None,
         self.target_files_factor = target_files_factor
         self.max_commits = max_commits
 
-        # TODO: pass through command line
-        self.github_handler.tokens = tokens
         # TODO: other scenarios should extend fix scenario
         fix_scn = self.sampling_handler.get_scenario(dataset, scenario='fix', extension=self.extensions)
         self.app.log.info(f"Generating fix scenario...")
diff --git a/tenet/plugins/hybrid.py b/tenet/plugins/hybrid.py
index 6508b5f..fe54dc2 100644
--- a/tenet/plugins/hybrid.py
+++ b/tenet/plugins/hybrid.py
@@ -15,6 +15,12 @@ class Meta:
     def __init__(self, **kw):
         super().__init__(**kw)
 
+    def set_sources(self):
+        pass
+
+    def get_sinks(self):
+        pass
+
     def run(self, dataset: pd.DataFrame, inner_join: bool = False, **kwargs) -> Union[pd.DataFrame, None]:
         """
             runs the plugin
diff --git a/tenet/plugins/jscodeshift.py b/tenet/plugins/jscodeshift.py
index 72e9fd3..7c4e304 100644
--- a/tenet/plugins/jscodeshift.py
+++ b/tenet/plugins/jscodeshift.py
@@ -5,6 +5,7 @@
 from typing import Union
 from tqdm import tqdm
 
+from tenet.core.exc import TenetError
 from tenet.data.diff import FunctionBoundary, InlineDiff
 from tenet.data.schema import ContainerCommand
 from tenet.handlers.plugin import PluginHandler
@@ -23,23 +24,23 @@ def __init__(self, **kw):
         self.fn_boundaries = None
         self.add_labelled_nulls = None
 
+    def set_sources(self):
+        self.set('dataset_path', self.output)
+        self.set('fn_boundaries_file', self.path / 'output.txt')
+
+    def get_sinks(self):
+        self.get('raw_files_path')
+
     def run(self, dataset: pd.DataFrame, image_name: str = "jscodeshift", single_unsafe_fn: bool = False,
             add_labelled_nulls: str = None, **kwargs) -> Union[pd.DataFrame, None]:
         """
             runs the plugin
         """
         self.add_labelled_nulls = add_labelled_nulls
-        self.fn_boundaries_file = (self.path / 'output.txt')
-        self.set('fn_boundaries_file', self.fn_boundaries_file)
-        self.set('dataset_path', self.output)
-
-        if not self.get('raw_files_path'):
-            self.app.log.warning(f"raw files path not instantiated.")
-            return None
 
-        raw_files_path = Path(str(self.get('raw_files_path')).replace(str(self.app.workdir), str(self.app.bind)))
+        raw_files_path = Path(str(self.sinks['raw_files_path']).replace(str(self.app.workdir), str(self.app.bind)))
 
-        if not self.fn_boundaries_file.exists():
+        if not self.sources['fn_boundaries_file'].exists():
             # TODO: fix the node name
             container = self.container_handler.run(image_name=image_name)
             cmd = ContainerCommand(org=f"jscodeshift -p -s -d -t /js-fn-rearrange/transforms/outputFnBoundary.js {raw_files_path}")
@@ -47,20 +48,18 @@ def run(self, dataset: pd.DataFrame, image_name: str = "jscodeshift", single_uns
             self.container_handler.stop(container)
 
         try:
-            if not self.fn_boundaries_file.exists():
-                self.app.log.error(f"jscodeshift output file {self.fn_boundaries_file} not found")
-                return None
+            if not self.sources['fn_boundaries_file'].exists():
+                raise TenetError(f"jscodeshift output file {self.sources['fn_boundaries_file']} not found")
 
-            if not self.fn_boundaries_file.stat().st_size > 0:
-                self.app.log.error(f"jscodeshift output file {self.fn_boundaries_file} is empty")
-                return
+            if not self.sources['fn_boundaries_file'].stat().st_size > 0:
+                raise TenetError(f"jscodeshift output file {self.sources['fn_boundaries_file']} is empty")
 
         except TypeError as te:
             self.app.log.error(te)
             self.app.log.warning(f"jscodeshift output file not instantiated.")
             return None
 
-        outputs = self.fn_boundaries_file.open(mode='r').readlines()
+        outputs = self.sources['fn_boundaries_file'].open(mode='r').readlines()
         self.fn_boundaries = {}
         raw_files_path = str(raw_files_path).replace(str(self.app.workdir), str(self.app.bind))
 
diff --git a/tenet/plugins/jsonifier.py b/tenet/plugins/jsonifier.py
index b513340..d91f068 100644
--- a/tenet/plugins/jsonifier.py
+++ b/tenet/plugins/jsonifier.py
@@ -18,55 +18,45 @@ class Meta:
 
     def __init__(self, **kw):
         super().__init__(**kw)
+        self.jsonlines_files = {}
+        self.offset_files = {}
+
+    def set_sources(self):
+        for name, file in self.sinks.items():
+            if name == 'raw_files_path':
+                continue
+            jsonlines_path = f'{file.stem}_lines'
+            self.jsonlines_files[jsonlines_path] = Path(self.path, f'{file.stem}.jsonl')
+            self.set(jsonlines_path, self.jsonlines_files[jsonlines_path])
+
+        if 'offset' in self.node.kwargs and self.node.kwargs['offset']:
+            for name, file in self.sinks.values():
+                if name == 'raw_files_path':
+                    continue
+                offset_path = f'{file.stem}_offset_path'
+                self.offset_files[offset_path] = Path(self.path, f'{file.stem}_offset_dict.json')
+                self.set(offset_path, self.offset_files[offset_path])
+
+    def get_sinks(self):
+        self.get('train_data_path')
+        self.get('val_data_path')
+        self.get('test_data_path')
+        self.get('raw_files_path')
 
     def run(self, dataset: pd.DataFrame, offset: bool = False, **kwargs) -> Union[pd.DataFrame, None]:
         """
             runs the plugin
         """
 
-        train_data_path = self.get('train_data_path')
-        val_data_path = self.get('val_data_path')
-        test_data_path = self.get('test_data_path')
-        raw_files_path_str = self.get('raw_files_path')
-
-        if not train_data_path:
-            self.app.log.error(f"Train data path not instantiated")
-            return None
-
-        if not val_data_path:
-            self.app.log.error(f"Val data path not instantiated")
-            return None
-
-        if not test_data_path:
-            self.app.log.error(f"Test data path not instantiated")
-            return None
-
-        if not raw_files_path_str:
-            self.app.log.error(f"Raw files path not instantiated")
-            return None
-
-        raw_files_path = Path(raw_files_path_str)
+        raw_files_path = Path(self.sinks['raw_files_path_str'])
 
         if not raw_files_path.exists():
             self.app.log.warning(f"{raw_files_path} not found")
             return None
 
-        split_files = [train_data_path, val_data_path, test_data_path]
-        jsonlines_files = {}
-        offset_files = {}
-
-        for file in split_files:
-            jsonlines_path = f'{file.stem}_lines'
-            jsonlines_files[jsonlines_path] = Path(self.path, f'{file.stem}.jsonl')
-            self.set(jsonlines_path, jsonlines_files[jsonlines_path])
-
-        if offset:
-            for file in split_files:
-                offset_path = f'{file.stem}_offset_path'
-                offset_files[offset_path] = Path(self.path, f'{file.stem}_offset_dict.json')
-                self.set(offset_path, offset_files[offset_path])
-
-        for jsonlines_file, split_file in zip(jsonlines_files.values(), split_files):
+        for jsonlines_file, (name, split_file) in zip(self.jsonlines_files.values(), self.sinks.items()):
+            if name == 'raw_files_path':
+                continue
             self.app.log.info(f"======== Building JSONLines for {split_file}... ========")
 
             funcs_df = pd.read_csv(split_file)
@@ -79,7 +69,7 @@ def run(self, dataset: pd.DataFrame, offset: bool = False, **kwargs) -> Union[pd
                     output_file.write({'code': row.code, 'label': row.label})
 
         if offset:
-            for jsonlines_file, offset_file in zip(jsonlines_files.values(), offset_files.values()):
+            for jsonlines_file, offset_file in zip(self.jsonlines_files.values(), self.offset_files.values()):
                 to_offset(jsonlines_file, offset_file)
 
         return dataset
diff --git a/tenet/plugins/labeler.py b/tenet/plugins/labeler.py
index e13e032..bfbd41f 100644
--- a/tenet/plugins/labeler.py
+++ b/tenet/plugins/labeler.py
@@ -5,6 +5,7 @@
 from typing import Union
 
 from tenet.core.diff_labeller.labeler import Labeler as DiffLabeler
+from tenet.core.exc import TenetError
 from tenet.data.diff import Entry, DiffBlock
 from tenet.handlers.plugin import PluginHandler
 
@@ -23,8 +24,11 @@ def __init__(self, **kw):
         self.inline_dir = None
         self.sim_ratio_thresh = None
 
-    def set_dirs(self):
-        self.inline_dir = self.output.parent / 'inline'
+    def set_sources(self):
+        pass
+
+    def get_sinks(self):
+        self.get('raw_files_path')
 
     def get_file_str(self, file: Path) -> str:
         """
@@ -45,17 +49,8 @@ def run(self, dataset: pd.DataFrame, file_size_limit: int = None, sim_ratio_tres
         """
             runs the plugin
         """
-        if not self.get('raw_files_path'):
-            self.app.log.warning(f"Raw files path not instantiated.")
-            return None
-
-        if not self.get('raw_files_path').exists():
-            self.app.log.warning(f"Train data file not found.")
-            return None
-
         self.sim_ratio_thresh = sim_ratio_tresh
-        self.set('dataset', str(self.output))
-        self.set_dirs()
+        self.inline_dir = self.output.parent / 'inline'
 
         if file_size_limit:
             self.file_size_limit = file_size_limit
@@ -80,8 +75,8 @@ def to_inline(self, entry: Entry) -> list:
         inline_proj_dir = self.inline_dir / f"{entry.owner}_{entry.project}_{entry.a_version}_{entry.b_version}"
 
         try:
-            a_str = self.get_file_str(file=self.get('raw_files_path') / entry.full_a_path)
-            b_str = self.get_file_str(file=self.get('raw_files_path') / entry.full_b_path)
+            a_str = self.get_file_str(file=self.sinks['raw_files_path'] / entry.full_a_path)
+            b_str = self.get_file_str(file=self.sinks['raw_files_path'] / entry.full_b_path)
             # Perform pretty-printing and diff comparison
             labeler = DiffLabeler(entry=entry, a_str=a_str, b_str=b_str, inline_proj_dir=inline_proj_dir)
             labeler(unsafe_label=entry.label)
diff --git a/tenet/plugins/merge_sources.py b/tenet/plugins/merge_sources.py
deleted file mode 100644
index bc73faa..0000000
--- a/tenet/plugins/merge_sources.py
+++ /dev/null
@@ -1,35 +0,0 @@
-import pandas as pd
-
-from typing import Union
-
-from tenet.handlers.plugin import PluginHandler
-
-
-class MergeSources(PluginHandler):
-    """
-        MergeSources plugin
-    """
-
-    class Meta:
-        label = "merge_sources"
-
-    def run(self, dataset: pd.DataFrame, **kwargs) -> Union[pd.DataFrame, None]:
-        """
-            runs the plugin
-        """
-        sources = [connector.source for connector in self.get_sinks().values() if 'dataset' in connector.links]
-
-        if sources:
-            dataset_paths = [self.app.executed_edges[s].output for s in sources if s in self.app.executed_edges]
-            datasets = [pd.read_csv(d, index_col=False) for d in dataset_paths]
-
-            if datasets:
-                df = pd.concat(datasets, ignore_index=True)
-                self.app.log.info(f"Total number of entries: {len(df)}")
-                return df
-
-        return None
-   
-
-def load(app):
-    app.handler.register(MergeSources)
diff --git a/tenet/plugins/ml_pipeline.py b/tenet/plugins/ml_pipeline.py
index 30efb17..805e144 100644
--- a/tenet/plugins/ml_pipeline.py
+++ b/tenet/plugins/ml_pipeline.py
@@ -22,29 +22,35 @@ class MLPipeline(PluginHandler):
     class Meta:
         label = "ml_pipeline"
 
+    def __init__(self, **kw):
+        super().__init__(**kw)
+        self.vectorizer = None
+
+    def set_sources(self):
+        self.vectorizer = str(self.sinks['train_features_path'].stem).split('_')[0]
+        model_path = self.path / f"{self.output.stem}_nlp_{self.vectorizer}_{self.node.kwargs['model_type']}.model"
+        self.set('model_path', model_path)
+
+    def get_sinks(self):
+        self.get('train_features_path')
+        self.get('train_labels_path')
+        self.get('test_features_path')
+        self.get('test_labels_path')
+        self.get('train_data_path')
+        self.get('test_data_path')
+
     def run(self, dataset: pd.DataFrame, model_type: str = 'RFC', training: bool = True, evaluate: bool = True,
             **kwargs) -> Union[pd.DataFrame, None]:
         """
             runs the plugin
         """
-        train_features_path = self.get('train_features_path')
-        train_labels_path = self.get('train_labels_path')
-        test_features_path = self.get('test_features_path')
-        test_labels_path = self.get('test_labels_path')
-        train_data_path = self.get('train_data_path')
-        test_data_path = self.get('test_data_path')
-
         # Set models path after data grouping options set.
-        vectorizer = str(train_features_path.stem).split('_')[0]
         model, model_pipeline, model_prefix, model_param_grid = get_model(model_type,
                                                                           n_jobs=self.app.get_config('local_threads'))
-        model_path = self.path / f"{self.output.stem}_nlp_{vectorizer}_{model_prefix}.model"
-        self.set('model_path', model_path)
-        predictions_path = self.path / f"{self.output.stem}_{model_prefix}_{vectorizer}_predictions.csv"
+        predictions_path = self.path / f"{self.output.stem}_{model_prefix}_{self.vectorizer}_predictions.csv"
         # Load data.
-        self.app.log.info(train_labels_path)
-        train_labels = pd.read_csv(train_labels_path)
-        test_labels = pd.read_csv(test_labels_path)
+        train_labels = pd.read_csv(self.sinks['train_labels_path'])
+        test_labels = pd.read_csv(self.sinks['test_labels_path'])
         y_train = self.convert_labels(train_labels["label"]).to_numpy().reshape((-1, 1))
         y_test = self.convert_labels(test_labels["label"]).to_numpy().reshape((-1, 1))
 
@@ -54,8 +60,8 @@ def run(self, dataset: pd.DataFrame, model_type: str = 'RFC', training: bool = T
         self.app.log.info(f"x_train shape: {x_train.shape}, x_test shape: {x_test.shape}")
 
         self.app.log.info("Adding NLP features.")
-        train_features_sparse = load_npz(train_features_path)
-        test_features_sparse = load_npz(test_features_path)
+        train_features_sparse = load_npz(self.sinks['train_features_path'])
+        test_features_sparse = load_npz(self.sinks['test_features_path'])
         x_train = hstack([x_train, train_features_sparse])
         x_test = hstack([x_test, test_features_sparse])
         self.app.log.info(f"x_train shape: {x_train.shape}, x_test shape: {x_test.shape}")
@@ -66,21 +72,21 @@ def run(self, dataset: pd.DataFrame, model_type: str = 'RFC', training: bool = T
         if training:
             # Tune, train and validate model.
             t_start = time.time()
-            self.tune_train_and_validate(model_path=model_path, model=model_pipeline, param_grid=model_param_grid,
-                                         x_train=x_train, y_train=y_train)
+            self.tune_train_and_validate(model_path=self.sources['model_path'], model=model_pipeline,
+                                         param_grid=model_param_grid, x_train=x_train, y_train=y_train)
             train_time = time.time() - t_start
 
             self.app.log.info(f"Training time {train_time}")
 
         if evaluate:
-            y_predict, results = self.evaluate_model(model_path=model_path, x_test=x_test, y_test=y_test)
-            res_df = pd.DataFrame([results])
-            res_df.to_csv(str(self.path / f'{model_prefix}_metrics.csv'))
-            test_data = pd.read_csv(str(test_data_path))
+            y_predict, results = self.evaluate_model(model_path=self.sources['model_path'], x_test=x_test, y_test=y_test)
+            test_data = pd.read_csv(str(self.sinks['test_data_path']))
             test_data['predicted'] = y_predict.tolist()
             test_data.to_csv(str(predictions_path), index=False)
+            results['model'] = model_type
+            return pd.DataFrame([results])
 
-        return None
+        return pd.DataFrame([])
 
     def tune_train_and_validate(self, model_path: Path, model: Any, param_grid: List[dict], x_train: Any, y_train: Any):
         # Perform gridsearch.
diff --git a/tenet/plugins/nvd_prepare.py b/tenet/plugins/nvd_prepare.py
index 664b1cf..a49458b 100644
--- a/tenet/plugins/nvd_prepare.py
+++ b/tenet/plugins/nvd_prepare.py
@@ -18,20 +18,21 @@ class NVDPrepare(PluginHandler):
     class Meta:
         label = "nvd_prepare"
 
-    def run(self, dataset: pd.DataFrame, tokens: Union[str, list] = None, metadata: bool = True, language: bool = True,
-            extension: bool = True, include_comments: bool = True, drop_patch: bool = True, **kwargs) \
+    def set_sources(self):
+        self.set('metadata_path', self.path / 'metadata')
+        self.set('normalized_path', self.path / f'{self.output.stem}_normalized.csv')
+
+    def get_sinks(self):
+        pass
+
+    def run(self, dataset: pd.DataFrame, metadata: bool = True, language: bool = True, extension: bool = True,
+            include_comments: bool = True, drop_patch: bool = True, **kwargs) \
             -> Union[pd.DataFrame, None]:
         """
             runs the plugin
         """
-        metadata_path = self.path / 'metadata'
-        #f'{self.output.stem}_metadata.csv'
-        self.set('metadata_path', metadata_path)
-        df_normalized_path = self.path / f'{self.output.stem}_normalized.csv'
-        self.set('normalized_path', df_normalized_path)
-        self.github_handler.tokens = tokens
-
-        if not df_normalized_path.exists():
+
+        if not self.sources['normalized_path'].exists():
             dataset.rename(inplace=True, columns={'cve_id': 'vuln_id', 'cwes': 'cwe_id', 'commits': 'chain',
                                                   'description': 'summary', 'impact': 'score'})
             dataset = dataset[['vuln_id', 'cwe_id', 'score', 'chain', 'summary', 'published_date']]
@@ -52,9 +53,9 @@ def run(self, dataset: pd.DataFrame, tokens: Union[str, list] = None, metadata:
             self.app.log.info(f"Entries (after nan drop): {len(dataset)}")
 
             dataset = transform_to_commits(dataset)
-            dataset.to_csv(str(df_normalized_path))
+            dataset.to_csv(str(self.sources['normalized_path']))
         else:
-            dataset = pd.read_csv(str(df_normalized_path))
+            dataset = pd.read_csv(str(self.sources['normalized_path']))
 
         self.app.log.info(f"Size after normalization: {len(dataset)}")
 
@@ -63,9 +64,10 @@ def run(self, dataset: pd.DataFrame, tokens: Union[str, list] = None, metadata:
             for project, rows in tqdm(dataset.groupby(['project'])):
                 self.multi_task_handler.add(project=project, chains=rows['chain'].to_list(), indexes=rows.index,
                                             include_comments=include_comments, commits=rows['commit_sha'].to_list(),
-                                            save_path=metadata_path, drop_patch=drop_patch)
+                                            save_path=self.sources['metadata_path'], drop_patch=drop_patch)
 
             self.multi_task_handler(func=self.github_handler.get_project_metadata)
+
             try:
                 metadata_df = pd.concat(self.multi_task_handler.results())
             except ValueError as ve:
diff --git a/tenet/plugins/nvd_source.py b/tenet/plugins/nvd_source.py
index eb9ea7c..1e98b53 100644
--- a/tenet/plugins/nvd_source.py
+++ b/tenet/plugins/nvd_source.py
@@ -21,6 +21,12 @@ class NVDSource(PluginHandler):
     class Meta:
         label = "nvd_source"
 
+    def set_sources(self):
+        pass
+
+    def get_sinks(self):
+        pass
+
     def plot(self, dataset: pd.DataFrame, **kwargs):
         """ Print commits statistics. """
 
diff --git a/tenet/plugins/osv_prepare.py b/tenet/plugins/osv_prepare.py
index 2735123..756f635 100644
--- a/tenet/plugins/osv_prepare.py
+++ b/tenet/plugins/osv_prepare.py
@@ -18,25 +18,30 @@ class OSVprepare(PluginHandler):
     class Meta:
         label = "osv_prepare"
 
-    def run(self, dataset: pd.DataFrame, tokens: Union[str, list] = None, metadata: bool = True, language: bool = True,
-            extension: bool = True, include_comments: bool = True, drop_patch: bool = True, **kwargs) \
+    def __init__(self, **kw):
+        super().__init__(**kw)
+
+    def set_sources(self):
+        self.set('metadata_path', self.path / 'metadata')
+        self.set('normalized_path', self.path / f'{self.output.stem}_normalized.csv')
+
+    def get_sinks(self):
+        pass
+
+    def run(self, dataset: pd.DataFrame, metadata: bool = True, language: bool = True, extension: bool = True,
+            include_comments: bool = True, drop_patch: bool = True, **kwargs) \
             -> Union[pd.DataFrame, None]:
         """
             runs the plugin
         """
-        metadata_path = self.path / 'metadata'
-        #f'{self.output.stem}_metadata.csv'
-        self.set('metadata_path', metadata_path)
-        df_normalized_path = self.path / f'{self.output.stem}_normalized.csv'
-        self.set('normalized_path', df_normalized_path)
-        self.github_handler.tokens = tokens
-
-        if not df_normalized_path.exists():
+
+        if not self.sources['normalized_path'].exists():
             dataset = dataset.rename(columns={'commits': 'chain'})
             dataset['summary'] = dataset.apply(lambda x: join(x['summary'], x['details']), axis=1)
             dataset = dataset[['vuln_id', 'cwe_id', 'score', 'chain', 'summary', 'published_date']]
             dataset['dataset'] = "OSV"
             dataset = self.normalize(dataset)
+
             for idx, row in tqdm(dataset.iterrows()):
                 self.multi_task_handler.add(chain=row['chain']).update_id(idx)
 
@@ -50,9 +55,9 @@ def run(self, dataset: pd.DataFrame, tokens: Union[str, list] = None, metadata:
             self.app.log.info(f"Entries (after nan drop): {len(dataset)}")
 
             dataset = transform_to_commits(dataset)
-            dataset.to_csv(str(df_normalized_path))
+            dataset.to_csv(str(self.sources['normalized_path']))
         else:
-            dataset = pd.read_csv(str(df_normalized_path))
+            dataset = pd.read_csv(str(self.sources['normalized_path']))
 
         self.app.log.info(f"Size after normalization: {len(dataset)}")
 
@@ -61,9 +66,10 @@ def run(self, dataset: pd.DataFrame, tokens: Union[str, list] = None, metadata:
             for project, rows in tqdm(dataset.groupby(['project'])):
                 self.multi_task_handler.add(project=project, chains=rows['chain'].to_list(), indexes=rows.index,
                                             include_comments=include_comments, commits=rows['commit_sha'].to_list(),
-                                            save_path=metadata_path, drop_patch=drop_patch)
+                                            save_path=self.sources['metadata_path'], drop_patch=drop_patch)
 
             self.multi_task_handler(func=self.github_handler.get_project_metadata)
+
             try:
                 metadata_df = pd.concat(self.multi_task_handler.results())
             except ValueError as ve:
diff --git a/tenet/plugins/osv_source.py b/tenet/plugins/osv_source.py
index 838c679..0b0cf8a 100644
--- a/tenet/plugins/osv_source.py
+++ b/tenet/plugins/osv_source.py
@@ -26,6 +26,12 @@ class OSVSource(PluginHandler):
     class Meta:
         label = "osv_source"
 
+    def set_sources(self):
+        pass
+
+    def get_sinks(self):
+        pass
+
     def plot(self, dataset: pd.DataFrame, **kwargs):
         """ Print commits statistics. """
 
@@ -58,7 +64,7 @@ def run(self,
             runs the plugin
         """
 
-        token, df_ecosystems = kwargs['tokens'][-1], []
+        df_ecosystems = []
         # set folder for GHSA dump
         ecosystem = 'GHSA'
         ghsa_metadata_path = Path(self.path / ecosystem)
@@ -67,7 +73,7 @@ def run(self,
             ghsa_metadata_path.mkdir()
         # get GitHub advisories (GHSA)
         self.app.log.info(f"Downloading {ecosystem} metadata...")
-        dump(ghsa_metadata_path, token)
+        dump(ghsa_metadata_path, self.app.pargs.tokens.split(',')[0])
 
         # df_metadata_path = self.path / kwargs['dataset_name']
         ghsa_results = self.parse_json_ghsa(fin_path=ghsa_metadata_path)
diff --git a/tenet/plugins/preprocess.py b/tenet/plugins/preprocess.py
index bc25cb9..87ef180 100644
--- a/tenet/plugins/preprocess.py
+++ b/tenet/plugins/preprocess.py
@@ -19,6 +19,17 @@ class Meta:
     def __init__(self, **kw):
         super().__init__(**kw)
 
+    def set_sources(self):
+        self.set('train_output_path', self.path / f'{self.output.stem}.train.c2v')
+        self.set('val_output_path', self.path / f'{self.output.stem}.val.c2v')
+        self.set('test_output_path', self.path / f'{self.output.stem}.test.c2v')
+        self.set('dict_file_path', self.path / f'{self.output.stem}.dict.c2v')
+
+    def get_sinks(self):
+        self.get('train_data_path')
+        self.get('val_data_path')
+        self.get('test_data_path')
+
     def run(self, dataset: pd.DataFrame, max_contexts: int = 200, word_vocab_size: int = 1301136,
             path_vocab_size: int = 911417, target_vocab_size: int = 261245, **kwargs) -> Union[pd.DataFrame, None]:
         """
@@ -29,47 +40,9 @@ def run(self, dataset: pd.DataFrame, max_contexts: int = 200, word_vocab_size: i
                 target_vocab_size (int): Max number of target words to keep in the vocabulary
         """
 
-        dataset_name = self.output.stem
-        train_output_path = f'{self.path}/{dataset_name}.train.c2v'
-        self.set('train_output_path', train_output_path)
-        val_output_path = f'{self.path}/{dataset_name}.val.c2v'
-        self.set('val_output_path', val_output_path)
-        test_output_path = f'{self.path}/{dataset_name}.test.c2v'
-        self.set('test_output_path', test_output_path)
-        dict_file_path = Path(self.path, f'{dataset_name}.dict.c2v')
-        self.set('dict_file_path', dict_file_path)
-
-        train_data_path = self.get('train_data_path')
-        val_data_path = self.get('val_data_path')
-        test_data_path = self.get('test_data_path')
-
-        if not train_data_path:
-            self.app.log.warning(f"Train data file not instantiated.")
-            return None
-
-        if not Path(train_data_path).exists():
-            self.app.log.warning(f"Train data file not found.")
-            return None
-
-        if not val_data_path:
-            self.app.log.warning(f"Validation data file not instantiated.")
-            return None
-
-        if not Path(val_data_path).exists():
-            self.app.log.warning(f"Validation data file not found.")
-            return None
-
-        if not test_data_path:
-            self.app.log.warning(f"Test data file not instantiated.")
-            return None
-
-        if not Path(test_data_path).exists():
-            self.app.log.warning(f"Test data file not found.")
-            return None
-
-        train_data = pd.read_csv(str(train_data_path))
-        test_data = pd.read_csv(str(test_data_path))
-        val_data = pd.read_csv(str(val_data_path))
+        train_data = pd.read_csv(str(self.sinks['train_data_path']))
+        test_data = pd.read_csv(str(self.sinks['test_data_path']))
+        val_data = pd.read_csv(str(self.sinks['val_data_path']))
 
         self.app.log.info("Creating histogram from training data")
         word_vocab, path_vocab = get_histograms(train_data['context_paths'].to_list())
@@ -80,32 +53,36 @@ def run(self, dataset: pd.DataFrame, max_contexts: int = 200, word_vocab_size: i
                                              path_to_count=path_to_count)
         self.app.log.info("Truncating train data")
         train_context_paths, train_data_info = cp_truncator(context_paths=train_data['context_paths'].to_list(),
-                                                            labels=train_data['label'].to_list(), path=train_output_path)
+                                                            labels=train_data['label'].to_list(),
+                                                            path=self.sources['train_output_path'])
         self.app.log.info(f'Train data info: {train_data_info}')
 
-        with Path(train_output_path).open(mode='w') as f:
+        with Path(self.sources['train_output_path']).open(mode='w') as f:
             for cp in train_context_paths:
                 f.write(f"{cp}\n")
 
-        save_dictionaries(dict_file_path=dict_file_path, word_to_count=word_to_count, path_to_count=path_to_count,
-                          target_to_count=target_to_count, num_training_examples=train_data_info.total)
+        save_dictionaries(dict_file_path=self.sources['dict_file_path'], word_to_count=word_to_count,
+                          path_to_count=path_to_count, target_to_count=target_to_count,
+                          num_training_examples=train_data_info.total)
 
         self.app.log.info("Truncating test data")
         test_context_paths, test_data_info = cp_truncator(context_paths=test_data['context_paths'].to_list(),
-                                                          labels=test_data['label'].to_list(), path=test_output_path)
+                                                          labels=test_data['label'].to_list(),
+                                                          path=self.sources['test_output_path'])
         self.app.log.info(f'Test data info: {test_data_info}')
 
-        with Path(test_output_path).open(mode='w') as f:
+        with Path(self.sources['test_output_path']).open(mode='w') as f:
             for cp in test_context_paths:
                 f.write(f"{cp}\n")
 
         self.app.log.info("Truncating validation data")
         val_context_paths, val_data_info = cp_truncator(context_paths=val_data['context_paths'].to_list(),
-                                                        labels=val_data['label'].to_list(), path=val_output_path)
+                                                        labels=val_data['label'].to_list(),
+                                                        path=self.sources['val_output_path'])
         self.app.log.info(f'Val data info: {val_data_info}')
 
         return dataset
 
 
 def load(app):
-    app.handler.register(PreprocessHandler)
\ No newline at end of file
+    app.handler.register(PreprocessHandler)
diff --git a/tenet/plugins/sampling.py b/tenet/plugins/sampling.py
index 2944d63..a5bebf5 100644
--- a/tenet/plugins/sampling.py
+++ b/tenet/plugins/sampling.py
@@ -20,6 +20,14 @@ def __init__(self, **kw):
         self.token = None
         self.balance_techniques = ['stratified_pair_hash', 'stratified_column']
 
+    def set_sources(self):
+        self.set('train_data', self.path / 'train.csv')
+        self.set('val_data', self.path / 'val.csv')
+        self.set('test_data', self.path / 'test.csv')
+
+    def get_sinks(self):
+        pass
+
     def __call__(self, technique: str, seed: int, dataset: pd.DataFrame, undersample_safe: float = None,
                  column: str = None) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
         # TODO: adapt rest of techniques
@@ -64,14 +72,6 @@ def run(self, dataset: pd.DataFrame, technique: str = "", seed: int = 0, only_si
             :param only_single: flag for considering only single unsafe functions in a file
             :param only_multiple: flag for considering only multiple unsafe functions in a file
         """
-        # TODO: change these sets into something simpler
-        train_data_path = Path(self.path, 'train.csv')
-        val_data_path = Path(self.path, 'val.csv')
-        test_data_path = Path(self.path, 'test.csv')
-
-        self.set('train_data', train_data_path)
-        self.set('val_data', val_data_path)
-        self.set('test_data', test_data_path)
 
         if technique and technique not in self.balance_techniques:
             self.app.log.warning(f"Technique must be one of the following: {self.balance_techniques}")
@@ -124,12 +124,12 @@ def run(self, dataset: pd.DataFrame, technique: str = "", seed: int = 0, only_si
                           f"Val.: {len(val)} ({val.label.value_counts()})\n"
                           f"Test: {len(test)} ({test.label.value_counts()})")
 
-        train.to_csv(str(train_data_path))
-        val.to_csv(str(val_data_path))
-        test.to_csv(str(test_data_path))
+        train.to_csv(str(self.sources['train_data']))
+        val.to_csv(str(self.sources['val_data']))
+        test.to_csv(str(self.sources['test_data']))
 
         return dataset
 
 
 def load(app):
-    app.handler.register(SamplingHandler)
\ No newline at end of file
+    app.handler.register(SamplingHandler)
diff --git a/tenet/plugins/transform.py b/tenet/plugins/transform.py
index aacda43..9de7533 100644
--- a/tenet/plugins/transform.py
+++ b/tenet/plugins/transform.py
@@ -27,9 +27,13 @@ def __init__(self, **kw):
             snippet (bool): flag to transform samples to snippet granularity
         """
         super().__init__(**kw)
-        self.files_path = None
         self.fix = False
-        self.clean_files_dir: Path = None
+
+    def set_sources(self):
+        self.set('clean_files_path', Path(self.path, 'files'))
+
+    def get_sinks(self):
+        self.get('files_path')
 
     def run(self, dataset: pd.DataFrame, fix: bool = False) -> Union[pd.DataFrame, None]:
         """
@@ -38,10 +42,7 @@ def run(self, dataset: pd.DataFrame, fix: bool = False) -> Union[pd.DataFrame, N
             fix (bool): flag to add fix file
             snippet (bool): flag to transform samples to snippet granularity
         """
-        self.files_path = self.get('files_path')
-        self.clean_files_dir = Path(self.path, 'files')
-        check_or_create_dir(self.clean_files_dir)
-        self.set('clean_files_path', self.clean_files_dir)
+        check_or_create_dir(self.sources['clean_files_path'])
         self.fix = fix
 
         for row in tqdm(dataset.to_dict(orient='records')):
@@ -58,7 +59,7 @@ def run(self, dataset: pd.DataFrame, fix: bool = False) -> Union[pd.DataFrame, N
         return None
 
     def get_triplet(self, row: dict) -> Tuple[LocalGitFile, LocalGitFile, LocalGitFile]:
-        project_path = self.files_path / row['project_name']
+        project_path = self.sinks['files_path'] / row['project_name']
 
         if 'raw_url_vuln' in row and not pd.isnull(row['raw_url_vuln']):
             vuln_file = LocalGitFile(url=row['raw_url_vuln'], short=Path(row['file_path']), tag='vuln',
@@ -123,7 +124,7 @@ def get_clean_content(self, project: str, commit: str, file: LocalGitFile):
         if file is None:
             return None
 
-        clean_file = self.clean_files_dir / project / commit / file.short
+        clean_file = self.sources['clean_files_path'] / project / commit / file.short
 
         if clean_file.exists():
             return clean_file.open(mode='r').read()
diff --git a/tenet/utils/misc.py b/tenet/utils/misc.py
index 4ef6cc0..1f773ce 100644
--- a/tenet/utils/misc.py
+++ b/tenet/utils/misc.py
@@ -3,6 +3,9 @@
 import os
 import tarfile
 import json
+from typing import List
+
+import networkx as nx
 import numpy as np
 import pandas as pd