From 2ff33b2bf04b7c1f505ef420478d31a1a7c7d257 Mon Sep 17 00:00:00 2001 From: Ankita Katiyar Date: Fri, 17 Jan 2025 15:08:04 +0000 Subject: [PATCH] Add deployment related attributes Signed-off-by: Ankita Katiyar --- kedro/pipeline/pipeline.py | 40 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/kedro/pipeline/pipeline.py b/kedro/pipeline/pipeline.py index 826acd1b13..8f14bd0c14 100644 --- a/kedro/pipeline/pipeline.py +++ b/kedro/pipeline/pipeline.py @@ -369,6 +369,46 @@ def grouped_nodes(self) -> list[list[Node]]: return [list(group) for group in self._toposorted_groups] + @property + def grouped_by_namespace(self): + """Return a dictionary of the pipeline nodes grouped by namespace. + + Returns: + The pipeline nodes grouped by namespace. + """ + nodes_by_namespace = defaultdict(list) + for node in self.nodes: + if node.namespace: + nodes_by_namespace[node.namespace].append(node) + else: + nodes_by_namespace[node.name].append(node) + return nodes_by_namespace + + @property + def node_dependencies_by_namespace(self): + """Return a dictionary of the pipeline nodes dependencies grouped by namespace. + + Returns: + The pipeline nodes dependencies grouped by namespace. + """ + node_dependencies_by_namespace = defaultdict(dict) + for node in self.nodes: + key = node.namespace if node.namespace else node.name + for parent in self.node_dependencies[node]: + if key not in node_dependencies_by_namespace: + node_dependencies_by_namespace[key] = [] + if parent.namespace and parent.namespace != key: + node_dependencies_by_namespace[key].append(parent.namespace) + elif parent.namespace and parent.namespace == key: + continue + else: + node_dependencies_by_namespace[key].append(parent.name) + + node_dependencies_by_namespace = { + key: set(value) for key, value in node_dependencies_by_namespace.items() + } + return node_dependencies_by_namespace + def only_nodes(self, *node_names: str) -> Pipeline: """Create a new ``Pipeline`` which will contain only the specified nodes by name.