Source code for runway.cfngin.actions.graph

"""CFNgin graph action."""
import json
import logging
import sys

from ..plan import merge_graphs
from .base import BaseAction

LOGGER = logging.getLogger(__name__)

[docs]def each_step(graph): """Yield each step and it's direct dependencies. Args: graph (:class:`runway.cfngin.plan.Graph`): Graph to iterate over. Yields: Tuple[Step, Set(str)] """ steps = graph.topological_sort() steps.reverse() for step in steps: deps = graph.downstream( yield step, deps
[docs]def dot_format(out, graph, name="digraph"): """Output a graph using the graphviz "dot" format. Args: out (TextIo): Where output will be written. graph (:class:`runway.cfngin.plan.Graph`): Graph to be output. name (str): Name of the graph. """ out.write("digraph %s {\n" % name) for step, deps in each_step(graph): for dep in deps: out.write(' "%s" -> "%s";\n' % (step, dep)) out.write("}\n")
[docs]def json_format(out, graph): """Output the graph in a machine readable JSON format. Args: out (TextIo): Where output will be written. graph (:class:`runway.cfngin.plan.Graph`): Graph to be output. """ steps = {} for step, deps in each_step(graph): steps[] = {} steps[]["deps"] = [ for dep in deps] json.dump({"steps": steps}, out, indent=4) out.write("\n")
FORMATTERS = { "dot": dot_format, "json": json_format, }
[docs]class Action(BaseAction): """Responsible for outputing a graph for the current CFNgin config.""" DESCRIPTION = "Print graph" NAME = "graph" @property def _stack_action(self): """Run against a step.""" return None
[docs] def run(self, **kwargs): """Generate the underlying graph and prints it.""" graph = self._generate_plan( require_unlocked=False, include_persistent_graph=True ).graph if self.context.persistent_graph: graph = merge_graphs(self.context.persistent_graph, graph) if kwargs.get("reduce"): # This will performa a transitive reduction on the underlying # graph, producing less edges. Mostly useful for the "dot" format, # when converting to PNG, so it creates a prettier/cleaner # dependency graph. graph.transitive_reduction() fn = FORMATTERS[kwargs.get("format")] fn(sys.stdout, graph) sys.stdout.flush()