Create query for tracking metrics

package/kedro_viz/api/graphql/schema.py

@@ -17,9 +17,16 @@
 from kedro_viz import __version__
 from kedro_viz.data_access import data_access_manager
 from kedro_viz.integrations.pypi import get_latest_version, is_running_outdated_version
+from kedro_viz.models.experiment_tracking import TrackingDatasetGroup
 
-from .serializers import format_run, format_run_tracking_data, format_runs
+from .serializers import (
+    format_run,
+    format_run_metric_data,
+    format_run_tracking_data,
+    format_runs,
+)
 from .types import (
+    MetricDataset,
     Run,
     RunInput,
     TrackingDataset,
@@ -95,6 +102,26 @@ def run_tracking_data(
 
         return all_tracking_datasets
 
+    @strawberry.field(
+        description="Get metrics data for a limited number of recent runs"
+    )
+    def run_metrics_data(self, limit: Optional[int] = 25) -> MetricDataset:
+        run_ids = [
+            run.id for run in data_access_manager.runs.get_all_runs(limit_amount=limit)
+        ]
+        group = TrackingDatasetGroup.METRIC
+
+        metric_dataset_models = data_access_manager.tracking_datasets.get_tracking_datasets_by_group_by_run_ids(
+            run_ids, group
+        )
+
+        metric_data = {}
+        for dataset in metric_dataset_models:
+            metric_data[dataset.dataset_name] = dataset.runs
+
+        formatted_metric_data = format_run_metric_data(metric_data)
+        return MetricDataset(data=formatted_metric_data)
+
 
 @strawberry.type
 class Mutation:

package/kedro_viz/api/graphql/serializers.py

@@ -5,6 +5,7 @@
 
 import json
 from collections import defaultdict
+from itertools import product
 from typing import Dict, Iterable, List, Optional, cast
 
 from strawberry import ID
@@ -128,3 +129,51 @@ def format_run_tracking_data(
                 del formatted_tracking_data[tracking_key]
 
     return formatted_tracking_data
+
+
+def format_run_metric_data(metric_data: Dict) -> Dict:
+    """Format metric data to conforms to the schema required by plots on the front
+    end. Parallel Coordinate plots and Timeseries plots are supported.
+
+    Arguments:
+        metric_data: the data to format
+
+    Returns:
+        a dictionary containing the formatted metric data
+    """
+    formatted_metric_data = _initialise_metric_data_template(metric_data)
+    _populate_metric_data_template(metric_data, **formatted_metric_data)
+    return formatted_metric_data
+
+
+def _initialise_metric_data_template(metric_data: Dict) -> Dict:
+    """ """
+    runs, metrics = {}, {}
+    for dataset_name in metric_data:
+        dataset_name_root = dataset_name.rpartition(".")[0]
+        dataset = metric_data[dataset_name]
+        for run_id in dataset:
+            runs[run_id] = []
+            for metric in dataset[run_id]:
+                metrics[dataset_name_root + "." + metric] = []
+
+    for run_id in runs:
+        runs[run_id].extend([None] * len(metrics))
+    for metric in metrics:
+        metrics[metric].extend([None] * len(runs))
+
+    return {"metrics": metrics, "runs": runs}
+
+
+def _populate_metric_data_template(
+    metric_data: Dict, runs: Dict, metrics: Dict
+) -> None:
+    """ """
+    for (run_idx, run_id), (metric_idx, metric) in product(
+        enumerate(runs), enumerate(metrics)
+    ):
+        dataset_name_root, _, metric_name = metric.rpartition(".")
+        for dataset in metric_data:
+            if dataset_name_root == dataset.rpartition(".")[0]:
+                value = metric_data[dataset][run_id].get(metric_name, None)
+                runs[run_id][metric_idx] = metrics[metric][run_idx] = value

package/kedro_viz/api/graphql/types.py

@@ -34,6 +34,11 @@ class TrackingDataset:
     run_ids: List[ID]
 
 
+@strawberry.type(description="Metric data")
+class MetricDataset:
+    data: JSON
+
+
 TrackingDatasetGroup = strawberry.enum(
     TrackingDatasetGroupModel, description="Group to show kind of tracking data"
 )

package/kedro_viz/data_access/repositories/runs.py

@@ -43,13 +43,19 @@ def add_run(self, run: RunModel):
             session.add(run)
 
     @check_db_session
-    def get_all_runs(self) -> Optional[Iterable[RunModel]]:
+    def get_all_runs(
+        self, limit_amount: Optional[int] = None
+    ) -> Optional[Iterable[RunModel]]:
         all_runs = (
             self._db_session_class()  # type: ignore
             .query(RunModel)
             .order_by(RunModel.id.desc())
-            .all()
         )
+
+        if limit_amount:
+            all_runs = all_runs.limit(limit_amount)
+        all_runs = all_runs.all()
+
         if all_runs:
             self.last_run_id = all_runs[0].id
         return all_runs