def print_summary()

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

• 55 lines of code
• 15 McCabe index (conditional complexity)

    def print_summary(self, stream, log_file):

        # Early silent return if there are no queues, can happen
        # only in the case that the stream early returned due to
        # an inconsistent pipeline state.
        if not self._state.task_groups:
            return

        text = ""

        assert self._resolved_keys is not None
        elements = sorted(e for (e, k) in self._resolved_keys.items() if k != e._get_cache_key())
        if elements:
            text += self.content_profile.fmt("Resolved key Summary\n", bold=True)
            text += self.show_pipeline(elements, self.context.log_element_format)
            text += "\n\n"

        if self._failure_messages:
            values = {}

            for element_name, message in sorted(self._failure_messages.items()):
                for group in self._state.task_groups.values():
                    # Exclude the failure messages if the job didn't ultimately fail
                    # (e.g. succeeded on retry)
                    if element_name in group.failed_tasks:
                        values[element_name] = self._render(message)

            if values:
                text += self.content_profile.fmt("Failure Summary\n", bold=True)
                text += self._format_values(values, style_value=False)

        text += self.content_profile.fmt("Pipeline Summary\n", bold=True)
        values = {}

        values["Total"] = self.content_profile.fmt(str(len(stream.total_elements)))
        values["Session"] = self.content_profile.fmt(str(len(stream.session_elements)))

        processed_maxlen = 1
        skipped_maxlen = 1
        failed_maxlen = 1
        for group in self._state.task_groups.values():
            processed_maxlen = max(len(str(group.processed_tasks)), processed_maxlen)
            skipped_maxlen = max(len(str(group.skipped_tasks)), skipped_maxlen)
            failed_maxlen = max(len(str(len(group.failed_tasks))), failed_maxlen)

        for group in self._state.task_groups.values():
            processed = str(group.processed_tasks)
            skipped = str(group.skipped_tasks)
            failed = str(len(group.failed_tasks))

            processed_align = " " * (processed_maxlen - len(processed))
            skipped_align = " " * (skipped_maxlen - len(skipped))
            failed_align = " " * (failed_maxlen - len(failed))

            status_text = (
                self.content_profile.fmt("processed ")
                + self._success_profile.fmt(processed)
                + self.format_profile.fmt(", ")
                + processed_align
            )

            status_text += (
                self.content_profile.fmt("skipped ")
                + self.content_profile.fmt(skipped)
                + self.format_profile.fmt(", ")
                + skipped_align
            )

            status_text += self.content_profile.fmt("failed ") + self._err_profile.fmt(failed) + " " + failed_align
            values["{} Queue".format(group.name)] = status_text

        text += self._format_values(values, style_value=False)

        click.echo(text, nl=False, err=True)
        if log_file:
            click.echo(text, file=log_file, color=False, nl=False)