def show_pipeline()

in src/buildstream/_frontend/widget.py [0:0]

• 86 lines of code
• 30 McCabe index (conditional complexity)

    def show_pipeline(self, dependencies, format_):
        report = ""
        p = Profile()

        for element in dependencies:
            line = format_

            key = element._get_display_key()
            dim_keys = not key.strict

            # Guarantee that description is reported on a single line.
            description = " ".join(element._description.splitlines())

            line = p.fmt_subst(line, "name", element._get_full_name(), fg="blue", bold=True)
            line = p.fmt_subst(line, "key", key.brief, fg="yellow", dim=dim_keys)
            line = p.fmt_subst(line, "full-key", key.full, fg="yellow", dim=dim_keys)
            line = p.fmt_subst(line, "description", description, fg="yellow", dim=dim_keys)

            try:
                if not element._has_all_sources_resolved():
                    line = p.fmt_subst(line, "state", "no reference", fg="red")
                else:
                    if element.get_kind() == "junction":
                        line = p.fmt_subst(line, "state", "junction", fg="magenta")
                    elif not element._can_query_cache():
                        line = p.fmt_subst(line, "state", "waiting", fg="blue")
                    elif element._cached_failure():
                        line = p.fmt_subst(line, "state", "failed", fg="red")
                    elif element._cached_success():
                        line = p.fmt_subst(line, "state", "cached", fg="magenta")
                    elif not element._can_query_source_cache():
                        line = p.fmt_subst(line, "state", "waiting", fg="blue")
                    elif element._fetch_needed():
                        line = p.fmt_subst(line, "state", "fetch needed", fg="red")
                    elif element._buildable():
                        line = p.fmt_subst(line, "state", "buildable", fg="green")
                    else:
                        line = p.fmt_subst(line, "state", "waiting", fg="blue")
            except BstError as e:
                # Provide context to plugin error
                e.args = ("Failed to determine state for {}: {}".format(element._get_full_name(), str(e)),)
                raise e

            # Element configuration
            if "%{config" in format_:
                line = p.fmt_subst(
                    line,
                    "config",
                    _yaml.roundtrip_dump_string(element._Element__config),
                )

            # Variables
            if "%{vars" in format_:
                variables = dict(element._Element__variables)
                line = p.fmt_subst(line, "vars", _yaml.roundtrip_dump_string(variables))

            # Environment
            if "%{env" in format_:
                environment = element._Element__environment
                line = p.fmt_subst(line, "env", _yaml.roundtrip_dump_string(environment))

            # Public
            if "%{public" in format_:
                public = element._Element__public
                line = p.fmt_subst(line, "public", _yaml.roundtrip_dump_string(public))

            # Workspaced
            if "%{workspaced" in format_:
                line = p.fmt_subst(line, "workspaced", "(workspaced)" if element._get_workspace() else "", fg="yellow")

            # Workspace-dirs
            if "%{workspace-dirs" in format_:
                workspace = element._get_workspace()
                if workspace is not None:
                    path = workspace.get_absolute_path()
                    if path.startswith("~/"):
                        path = os.path.join(os.getenv("HOME", "/root"), path[2:])
                    line = p.fmt_subst(line, "workspace-dirs", "Workspace: {}".format(path))
                else:
                    line = p.fmt_subst(line, "workspace-dirs", "")

            # Dependencies
            if "%{deps" in format_:
                deps = [e._get_full_name() for e in element._dependencies(_Scope.ALL, recurse=False)]
                line = p.fmt_subst(line, "deps", _yaml.roundtrip_dump_string(deps).rstrip("\n"))

            # Build Dependencies
            if "%{build-deps" in format_:
                build_deps = [e._get_full_name() for e in element._dependencies(_Scope.BUILD, recurse=False)]
                line = p.fmt_subst(line, "build-deps", _yaml.roundtrip_dump_string(build_deps).rstrip("\n"))

            # Runtime Dependencies
            if "%{runtime-deps" in format_:
                runtime_deps = [e._get_full_name() for e in element._dependencies(_Scope.RUN, recurse=False)]
                line = p.fmt_subst(line, "runtime-deps", _yaml.roundtrip_dump_string(runtime_deps).rstrip("\n"))

            # Source Information
            if "%{source-info" in format_:

                # Get all the SourceInfo objects
                #
                all_source_infos = []
                for source in element.sources():
                    try:
                        source_infos = source.collect_source_info()
                    except SourceImplError as e:
                        source.warn(str(e))
                        continue

                    serialized_sources = []
                    for s in source_infos:
                        serialized = s._serialize()
                        serialized_sources.append(serialized)

                    all_source_infos += serialized_sources

                # Dump the SourceInfo provenance objects in yaml format
                line = p.fmt_subst(line, "source-info", _yaml.roundtrip_dump_string(all_source_infos))

            report += line + "\n"

        return report.rstrip("\n")