Add bst graph command

Adds a new command to bst called 'graph' which reimplements the same API
and functionality as contrib/bst-graph. The implementation details are
within _stream.py as another function called 'graph'.

Part of #1915.

Author: Arran Ireland

requirements/requirements.in

@@ -1,4 +1,5 @@
 Click >= 7.0
+graphviz
 grpcio
 Jinja2 >= 2.10
 pluginbase

requirements/requirements.txt

@@ -1,4 +1,5 @@
 click==8.1.7
+graphviz==0.20.3
 grpcio==1.65.1
 Jinja2==3.1.4
 pluginbase==1.0.1

src/buildstream/_frontend/cli.py

@@ -1654,3 +1654,29 @@ def artifact_delete(app, artifacts, deps):
     """Remove artifacts from the local cache"""
     with app.initialized():
         app.stream.artifact_delete(artifacts, selection=deps)
+
+
+###################################################################
+#                          Graph Command                          #
+###################################################################
+@cli.command(short_help="Render pipeline dependency graph.")
+@click.option(
+    "--format",
+    "-f",
+    "format_",
+    metavar="FORMAT",
+    default="dot",
+    type=click.STRING,
+    help="Render format: e.g. `dot`, `pdf`, `png`, `svg`, etc",
+)
+@click.option(
+    "--view",
+    is_flag=True,
+    help="Open the rendered graph with the default application",
+)
+@click.argument("element", nargs=1, required=True, type=click.Path())
+@click.pass_obj
+def graph(app, element, format_, view):
+    """Render dependency graph."""
+    with app.initialized():
+        app.stream.graph(element, format_, view)

src/buildstream/_stream.py

@@ -27,6 +27,7 @@
 from contextlib import contextmanager, suppress
 from collections import deque
 from typing import List, Tuple, Optional, Iterable, Callable
+from graphviz import Digraph
 
 from ._artifactelement import verify_artifact_ref, ArtifactElement
 from ._artifactproject import ArtifactProject
@@ -1228,6 +1229,39 @@ def redirect_element_names(self, elements):
 
         return list(output_elements)
 
+    # graph()
+    #
+    # Renders a dependency graph for the target element, in the given format and optionally opens it.
+    #
+    # Args:
+    #    target (str): The target element from which to build a dependency graph.
+    #
+    def graph(self, target, format_, view):
+        graph_ = Digraph()
+
+        for e in self.load_selection([target], selection=_PipelineSelection.ALL, need_state=False):
+            name = e._get_full_name()
+            build_deps = set(dep._get_full_name() for dep in e._dependencies(_Scope.BUILD, recurse=False) if dep)
+            runtime_deps = set(dep._get_full_name() for dep in e._dependencies(_Scope.RUN, recurse=False) if dep)
+
+            graph_.node(name)
+            for dep in build_deps:
+                graph_.edge(name, dep, label='build-dep')
+            for dep in runtime_deps:
+                graph_.edge(name, dep, label='runtime-dep')
+
+        graph_name = os.path.basename(target)
+        graph_name, _ = os.path.splitext(graph_name)
+        graph_path = graph_.render(cleanup=True,
+                     filename=graph_name,
+                     format=format_,
+                     view=view)
+
+        if graph_path:
+            self._context.messenger.info(f"Rendered dependency graph: {graph_path}")
+        else:
+            self._context.messenger.warn("Failed to render graph")
+
     # get_state()
     #
     # Get the State object owned by Stream