Source code for sceptre.config.graph

# -*- coding: utf-8 -*-


This module implements a StackGraph, which is represented as a directed
acyclic graph of a Stack's dependencies.

import logging
from typing import List

import networkx as nx
from sceptre.exceptions import CircularDependenciesError
from sceptre.stack import Stack

[docs]class StackGraph(object): """ A Directed Acyclic Graph representing the relationship between a Stack and its dependencies. Responsible for initalising the graph based on a set of Stacks. """ def __init__(self, stacks): """ Initialises a StackGraph based on a `set` of Stacks. :param stacks: A set of Stacks. :type stacks: set """ self.logger = logging.getLogger(__name__) self.graph = nx.DiGraph() self._generate_graph(stacks) def __repr__(self): return str(nx.convert.to_dict_of_lists(self.graph)) def __iter__(self): return self.graph.__iter__()
[docs] def filtered(self, source_stacks, reverse=False): graph = (nx.reverse if reverse else nx.DiGraph)(self.graph) relevant = set(source_stacks) for stack in source_stacks: relevant |= nx.algorithms.dag.ancestors(graph, stack) graph.remove_nodes_from({stack for stack in graph if stack not in relevant}) filtered = StackGraph(set()) filtered.graph = graph return filtered
[docs] def count_dependencies(self, stack): """ Returns the number of incoming edges a given Stack has in the StackGraph. The number of incoming edge also represents the number of Stacks that depend on the given Stack. """ return self.graph.in_degree(stack)
[docs] def remove_stack(self, stack): """ Removes a Stack from the StackGraph. This operation will also remove all adjecent edges that represent a 'depends on' relationship with other Stacks. """ return self.graph.remove_node(stack)
def _generate_graph(self, stacks): """ Generates the graph for the StackGraph object. :param stacks: A set of Stacks :type stacks: set """ for stack in stacks: self._generate_edges(stack, stack.dependencies) self.graph.remove_edges_from(nx.selfloop_edges(self.graph)) def _generate_edges(self, stack: Stack, dependencies: List[Stack]): """ Adds edges to the graph based on a list of dependencies that are generated from the initial stack config. Each of the paths in the inital_dependency_paths list are a depency that the inital Stack config depends on. :param stack: A Sceptre Stack :param dependencies: a collection of dependency paths """ self.logger.debug("Generate dependencies for stack {0}".format(stack)) for dependency in set(dependencies): self.graph.add_edge(dependency, stack) if not nx.is_directed_acyclic_graph(self.graph): raise CircularDependenciesError( f"Dependency cycle detected: {stack} {dependency}" ) self.logger.debug(" Added dependency: {}".format(dependency)) if not dependencies: self.graph.add_node(stack)