Merge pull request #412 from kikkomep/feat/stats

feat: workflows statistics

Author: kikkomep

lifemonitor/api/controllers.py

@@ -55,6 +55,42 @@ def _row_to_dict(row):
     return d
 
 
+@cached(timeout=Timeout.REQUEST)
+def stats_workflows_get():
+    """
+    Docstring for stats_get
+    """
+    workflows = None
+    if current_user and not current_user.is_anonymous:
+        workflows = lm.get_user_workflows(current_user, include_public=False,
+                                          include_subscriptions=True, only_subscriptions=True)
+    elif current_registry:
+        workflows = lm.get_registry_workflows(current_registry)
+    else:
+        workflows = lm.get_public_workflows()
+    stats = lm.get_workflows_stats(workflows)
+    logger.debug("stats_get. Got stats: %s", stats)
+    return stats
+
+
+@cached(timeout=Timeout.REQUEST)
+def stats_workflows_status_get():
+    """
+    Docstring for stats_get
+    """
+    workflows = None
+    if current_user and not current_user.is_anonymous:
+        workflows = lm.get_user_workflows(current_user, include_public=False,
+                                          include_subscriptions=True, only_subscriptions=True)
+    elif current_registry:
+        workflows = lm.get_registry_workflows(current_registry)
+    else:
+        workflows = lm.get_public_workflows()
+    stats = lm.get_workflows_status_stats(workflows)
+    logger.debug("stats_get. Got stats: %s", stats)
+    return stats
+
+
 @cached(timeout=Timeout.REQUEST)
 def workflow_registries_get():
     registries = lm.get_workflow_registries()
@@ -108,54 +144,54 @@ def workflow_registries_get_current():
     return lm_exceptions.report_problem(401, "Unauthorized")
 
 
-@cached(timeout=Timeout.REQUEST)
-def workflows_get(status=False, versions=False, subscriptions=False,
-                  page: int = 1, per_page: Optional[int] = None,
-                  max_items: Optional[int] = None,
-                  only_subscriptions=False,):
-    # Prepare pagination info
+def _get_workflows_list_(page: int = 1, per_page: Optional[int] = None, max_items: Optional[int] = None,
+                         subscriptions: bool = False, only_subscriptions: bool = False):
     page_info = PaginationInfo(page=page, per_page=per_page, max_items=max_items)
-    # Collect workflows depending on the current user/registry in session
     workflows = []
-    # If a user is logged in
     if current_user and not current_user.is_anonymous:
         logger.info("Getting workflows for user %s", current_user)
-        workflows.extend(lm.get_user_workflows(current_user,
-                                               include_subscriptions=subscriptions,
-                                               include_public=True,
-                                               only_subscriptions=only_subscriptions,
-                                               page=page_info))
-        logger.info("Total workflows for user %s: %d", current_user, len(workflows))
-    # If a registry is in the current session,
-    # get both public and registry workflows
+        workflows.extend(lm.get_user_workflows(
+            current_user,
+            include_subscriptions=subscriptions,
+            include_public=True,
+            only_subscriptions=only_subscriptions,
+            page=page_info
+        ))
     elif current_registry:
         logger.info("Getting workflows for registry %s", current_registry)
         workflows.extend(lm.get_registry_workflows(
             current_registry, include_public=True,
-            page=page_info))
-    # Engage public workflows only
+            page=page_info
+        ))
     else:
         logger.info("Getting public workflows")
         workflows.extend(lm.get_public_workflows(page=page_info))
+    return list(dict.fromkeys(workflows)), page_info
+
+
+@cached(timeout=Timeout.REQUEST)
+def workflows_get(status=False, versions=False, subscriptions=False,
+                  page: int = 1, per_page: Optional[int] = None,
+                  max_items: Optional[int] = None,
+                  only_subscriptions=False,
+                  stats: bool = False):
+    workflows, page_info = _get_workflows_list_(page, per_page, max_items,
+                                                subscriptions, only_subscriptions)
 
     logger.debug("workflows_get. Got %s workflows (user: %s)", len(workflows), current_user)
 
-    # Determine whose subscriptions to include
     subscriptions_of = []
     if subscriptions and current_user and not current_user.is_anonymous:
         subscriptions_of = [current_user]
 
-    # Create the serializer
     serializer = serializers.ListOfWorkflows(
         workflow_status=status,
         workflow_versions=versions,
         subscriptionsOf=subscriptions_of,
+        statistics=lm.get_workflows_stats(workflows) if stats else None,
         page=page_info
     )
-
-    # Remove duplicates and serialize the workflows
-    unique_workflows = list(dict.fromkeys(workflows))
-    return serializer.dump(unique_workflows)
+    return serializer.dump(workflows)
 
 
 def __get_workflow_version__(wf_uuid, wf_version=None) -> models.WorkflowVersion:
@@ -260,13 +296,14 @@ def workflows_rocrate_download(wf_uuid, wf_version):
 
 @authorized
 @cached(timeout=Timeout.REQUEST)
-def registry_workflows_get(status=False, versions=False,
+def registry_workflows_get(status=False, versions=False, stats=False,
                            page=None, per_page=None, max_items=None):
     page_info = PaginationInfo(page=page, per_page=per_page, max_items=max_items)
     workflows = lm.get_registry_workflows(current_registry, page=page_info)
     logger.debug("workflows_get. Got %s workflows (registry: %s)", len(workflows), current_registry)
     return serializers.ListOfWorkflows(
         workflow_status=status, workflow_versions=versions,
+        statistics=lm.get_workflows_stats(workflows) if stats else None,
         page=page_info
     ).dump(workflows)
 
@@ -282,6 +319,7 @@ def registry_workflows_post(body):
 @authorized
 @cached(timeout=Timeout.REQUEST)
 def registry_user_workflows_get(user_id, status=False, versions=False,
+                                stats=False,
                                 page=None, per_page=None, max_items=None):
     if not current_registry:
         return lm_exceptions.report_problem(401, "Unauthorized", detail=messages.no_registry_found)
@@ -291,7 +329,8 @@ def registry_user_workflows_get(user_id, status=False, versions=False,
         workflows = lm.get_user_registry_workflows(identity.user, current_registry, page=page_info)
         logger.debug("registry_user_workflows_get. Got %s workflows (user: %s)", len(workflows), current_user)
         return serializers.ListOfWorkflows(
-            workflow_status=status, workflow_versions=versions, page=page_info
+            workflow_status=status, workflow_versions=versions,
+            statistics=lm.get_workflows_stats(workflows) if stats else None, page=page_info
         ).dump(workflows)
     except OAuthIdentityNotFoundException as e:
         if logger.isEnabledFor(logging.DEBUG):
@@ -312,6 +351,7 @@ def registry_user_workflows_post(user_id, body):
 @authorized
 @cached(timeout=Timeout.REQUEST)
 def user_workflows_get(status=False, versions=False, subscriptions=False, only_subscriptions: bool = False,
+                       stats=False,
                        page=None, per_page=None, max_items=None):
     if not current_user or current_user.is_anonymous:
         return lm_exceptions.report_problem(401, "Unauthorized", detail=messages.no_user_in_session)
@@ -325,6 +365,7 @@ def user_workflows_get(status=False, versions=False, subscriptions=False, only_s
     return serializers.ListOfWorkflows(workflow_status=status,
                                        workflow_versions=versions,
                                        subscriptionsOf=[current_user] if subscriptions else None,
+                                       statistics=lm.get_workflows_stats(workflows) if stats else None,
                                        page=page_info).dump(workflows)
 
 

lifemonitor/api/serializers.py

@@ -21,9 +21,11 @@
 from __future__ import annotations
 
 import logging
-from typing import List
+from typing import List, Optional
 from urllib.parse import urljoin
 
+from marshmallow import fields, post_dump
+
 from lifemonitor import exceptions as lm_exceptions
 from lifemonitor import utils as lm_utils
 from lifemonitor.api.models.issues import WorkflowRepositoryIssue
@@ -32,7 +34,6 @@
 from lifemonitor.serializers import (BaseSchema, ListOfItems,
                                      ResourceMetadataSchema, ResourceSchema,
                                      ma)
-from marshmallow import fields, post_dump
 
 from . import models
 
@@ -525,13 +526,21 @@ class ListOfWorkflows(ListOfItems):
 
     subscriptionsOf: List[auth_models.User] = None
 
+    statistics = fields.Method("get_statistics")
+
     def __init__(self, *args,
                  workflow_status: bool = False, workflow_versions: bool = False,
-                 subscriptionsOf: List[auth_models.User] = None, **kwargs):
+                 subscriptionsOf: List[auth_models.User] = None,
+                 statistics: Optional[object] = None,
+                 **kwargs):
         super().__init__(*args, **kwargs)
         self.workflow_status = workflow_status
         self.workflow_versions = workflow_versions
         self.subscriptionsOf = subscriptionsOf
+        self._statistics = statistics
+
+    def get_statistics(self, workflow: models.Workflow):
+        return self._statistics if self._statistics else None
 
     def get_items(self, obj):
         exclude = ['meta', 'links']

lifemonitor/api/services.py

@@ -22,10 +22,11 @@
 
 import logging
 from datetime import datetime, timezone
-from typing import Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
 import lifemonitor.exceptions as lm_exceptions
 from lifemonitor.api import models
+from lifemonitor.api.models.status import AggregateTestStatus
 from lifemonitor.auth.models import (EventType,
                                      ExternalServiceAuthorizationHeader,
                                      HostingService, Notification, Permission,
@@ -822,3 +823,49 @@ def list_workflow_updates(since: datetime = datetime.now) -> Dict[str, datetime]
                 # "lastUpdate": int(row[2].timestamp() * 1000)
             })
         return result
+
+    @classmethod
+    def get_workflows_stats(cls, workflows: Optional[List[models.Workflow]] = None,
+                            status: bool = True) -> Dict[str, Any]:
+        # Reference to the list of workflows
+        workflows = workflows if workflows is not None else models.Workflow.all()
+        workflow_versions = [v for w in workflows for v in w.versions.values()]
+        test_suites = [s for wv in workflow_versions for s in wv.test_suites]
+        test_instances = [ti for ts in test_suites for ti in ts.test_instances]
+
+        # Basic stats
+        stats = {
+            "workflows": len(workflows),
+            "workflow_versions": len(workflow_versions),
+            "test_suites": len(test_suites),
+            "test_instances": len(test_instances),
+            "public_workflows": len([w for w in workflows if w.public]),
+            "private_workflows": len([w for w in workflows if not w.public]),
+        }
+
+        # Workflows status stats
+        if status:
+            stats["workflows_by_status"] = cls.get_workflows_status_stats(workflows)
+        return stats
+
+    @classmethod
+    def get_workflows_status_stats(cls, workflows: Optional[List[models.Workflow]] = None):
+        # Reference to the list of workflows
+        # Use provided workflows or all workflows if None
+        workflows = workflows if workflows is not None else models.Workflow.all()
+
+        # Group workflows by aggregated status
+        status_counts = {
+            AggregateTestStatus.ALL_PASSING: 0,
+            AggregateTestStatus.SOME_PASSING: 0,
+            AggregateTestStatus.ALL_FAILING: 0,
+            AggregateTestStatus.NOT_AVAILABLE: 0,
+        }
+
+        # Count workflows per status
+        for wf in workflows:
+            status = wf.latest_version.status.aggregated_status
+            if status not in status_counts:
+                status_counts[status] = 0
+            status_counts[status] += 1
+        return status_counts