feat: API pipeline run list has execution summary

[yuechao-qin]

TL;DR

Added execution summary to pipeline run responses to provide aggregated information about pipeline execution status (GET /api/pipeline_runs/)..

What changed?

• Added execution_summary field to PipelineRunResponse dataclass
• Renamed _get_execution_status_stats to _get_execution_stats_and_summary to return both stats and summary

How to test?

1. Create a pipeline run with multiple nodes in different states
2. Call the list API with include_execution_stats=True
3. Verify the response contains both execution stats and execution summary
4. Check that the summary correctly reports total nodes, ended nodes, and completion status

Run the new test cases in tests/test_api_server_sql.py:

pytest tests/test_api_server_sql.py

Why make this change?

This closes #87

[yuechao-qin]

Warning

This pull request is not mergeable via GitHub because a downstack PR is open. Once all requirements are satisfied, merge this PR as a stack on Graphite.
Learn more

• feat: GetExecutionNodes API returns number of total and ended nodes #99
• test: Add tests for GetGraphExecution API #98
• feat: API pipeline run get has execution summary #97
• feat: API pipeline run list has execution summary #96 👈 (View in Graphite)
• refactor: Function to get execution status stats #95
• feat: Model to track node states and has_ended flag #94
• master

This stack of pull requests is managed by Graphite. Learn more about stacking.

[morgan-wowk]

This is a pattern I would typically try to avoid in other programming languages. Here we are creating a an instance of the class with some initial state, then calling a function on that instance to later on to hydrate the instance with some values, to then return the instance to a caller.

What I would strive to due instead is have a pure / read only model that doesn't write to itself, or get written to after instatiation, but instead gets instantiated with its final values in the constructor. In this case, you could have a function build_execution_status_summar_from_stats that calculates the model attributes, instantiates the model, and returns the model in a read only state.

I could give examples of how this would look in other languages but that wouldn't be helpful.

Here is what Gemini says about implementing that in Python in a "pythonic" way:

@dataclasses.dataclass(frozen=True) # frozen=True makes instances immutable
class ExecutionStatusSummary:
    """A read-only summary of execution statuses."""
    total_executions: int
    ended_executions: int
    has_ended: bool

    @classmethod
    def from_stats(
        cls, stats: Dict[bts.ContainerExecutionStatus, int]
    ) -> "ExecutionStatusSummary":
        """
        Calculates summary attributes from a stats dictionary and returns a new
        immutable instance.
        """
        total_executions = sum(stats.values())

        ended_executions = sum(
            count
            for status, count in stats.items()
            if status in bts.CONTAINER_STATUSES_ENDED
        )

        # This logic is more robust: an empty set of executions hasn't "ended".
        has_ended = total_executions > 0 and ended_executions == total_executions

        return cls(
            total_executions=total_executions,
            ended_executions=ended_executions,
            has_ended=has_ended,
        )

and consuming it as:

summary = ExecutionStatusSummary.from_stats(stats)

[morgan-wowk]

This is not based on your exact changes and just an example ☝️ . In other words, not a copy/paste replacement and just a suggestion.

[Ark-kun]

I prefer less mutation too.
We've discussed this case with Yue Chao and the main reason for mutating implementation is because the instance will be constructed from multiple status stats objects.
Although, what we could do in that case is to first combine multiple dicts into one (via a new function or collection.Counter) and then initialize the structure.

full_status_stats = collections.Counter()
for status_stats in list_of_status_stats:
    full_status_stats.update(status_stats)
ExecutionStatusSummary.from_status_stats(full_status_stats)

[morgan-wowk]

https://app.graphite.com/github/pr/TangleML/tangle/96/feat-API-pipeline-run-list-has-execution-summary#comment-PRRC_kwDOOC2o5s6pgpDn

Could help simplify the code here