feat: calculate observation and license count from metrics

Author: StefanFl

backend/application/commons/migrations/0021_settings_observation_count_from_metrics.py

@@ -0,0 +1,18 @@
+# Generated by Django 6.0.3 on 2026-03-29 07:24
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("commons", "0020_settings_observation_title_notification_email_to_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="settings",
+            name="observation_count_from_metrics",
+            field=models.BooleanField(default=False),
+        ),
+    ]

backend/application/commons/migrations/0022_merge_20260330.py

@@ -0,0 +1,13 @@
+# Generated by Django 5.2.9 on 2026-03-13
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("commons", "0021_settings_observation_count_from_metrics"),
+        ("commons", "0022_merge_20260313"),
+    ]
+
+    operations = []

backend/application/commons/models.py

@@ -231,6 +231,8 @@ class Settings(Model, DirtyFieldsMixin):
         help_text="Time margin in seconds for checks of issued at, not before and expiration of OIDC tokens",
     )
 
+    observation_count_from_metrics = BooleanField(default=False)
+
     def save(self, *args: Any, **kwargs: Any) -> None:
         """
         Save object to the database. Removes all other entries if there

backend/application/core/api/views_product.py

@@ -28,6 +28,7 @@
 from application.access_control.queries.api_token import get_api_token_by_id
 from application.authorization.services.authorization import user_has_permission_or_403
 from application.authorization.services.roles_permissions import Permissions
+from application.commons.models import Settings
 from application.commons.services.log_message import format_log_message
 from application.core.api.filters import (
     BranchFilter,
@@ -118,8 +119,12 @@ class ProductGroupViewSet(ModelViewSet):
     search_fields = ["name"]
 
     def get_queryset(self) -> QuerySet[Product]:
-        return get_products(is_product_group=True, with_annotations=True)
-
+        settings = Settings.load()
+        return get_products(
+            is_product_group=True,
+            with_observation_annotations=not settings.observation_count_from_metrics,
+            with_metrics_annotations=settings.observation_count_from_metrics,
+        )
 
 class ProductGroupNameViewSet(GenericViewSet, ListModelMixin, RetrieveModelMixin):
     serializer_class = ProductNameSerializer
@@ -142,8 +147,13 @@ class ProductViewSet(ModelViewSet):
     search_fields = ["name"]
 
     def get_queryset(self) -> QuerySet[Product]:
+        settings = Settings.load()
         return (
-            get_products(is_product_group=False, with_annotations=True)
+            get_products(
+                is_product_group=False,
+                with_observation_annotations=not settings.observation_count_from_metrics,
+                with_metrics_annotations=settings.observation_count_from_metrics,
+            )
             .select_related("product_group")
             .select_related("product_group__license_policy")
             .select_related("repository_default_branch")

backend/application/core/migrations/0082_product_last_license_change.py

@@ -0,0 +1,19 @@
+# Generated by Django 6.0.3 on 2026-03-29 16:24
+
+import django.utils.timezone
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("core", "0081_product_observation_notification_min_priority_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="product",
+            name="last_license_change",
+            field=models.DateTimeField(default=django.utils.timezone.now),
+        ),
+    ]

backend/application/core/migrations/0083_merge_20260330.py

@@ -0,0 +1,13 @@
+# Generated by Django 5.2.9 on 2026-03-13
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("core", "0082_product_last_license_change"),
+        ("core", "0083_alter_observation_assessment_status_and_more"),
+    ]
+
+    operations = []

backend/application/core/models.py

@@ -105,6 +105,7 @@ class Product(Model, DirtyFieldsMixin):  # pylint: disable=too-many-instance-att
     issue_tracker_minimum_severity = CharField(max_length=12, choices=Severity.SEVERITY_CHOICES, blank=True)
 
     last_observation_change = DateTimeField(default=timezone.now)
+    last_license_change = DateTimeField(default=timezone.now)
 
     assessments_need_approval = BooleanField(default=False)
     new_observations_in_review = BooleanField(default=False)

backend/application/core/queries/product.py

@@ -1,6 +1,7 @@
+from datetime import date
 from typing import Optional
 
-from django.db.models import Count, Exists, IntegerField, OuterRef, Q, Subquery
+from django.db.models import Count, Exists, IntegerField, OuterRef, Q, Subquery, Sum
 from django.db.models.functions import Coalesce
 from django.db.models.query import QuerySet
 
@@ -15,17 +16,38 @@
 from application.core.types import Severity, Status
 from application.licenses.models import License_Component
 from application.licenses.types import License_Policy_Evaluation_Result
+from application.metrics.models import Product_License_Metrics, Product_Metrics
+
+SEVERITY_MAPPING = {
+    Severity.SEVERITY_CRITICAL: "active_critical",
+    Severity.SEVERITY_HIGH: "active_high",
+    Severity.SEVERITY_MEDIUM: "active_medium",
+    Severity.SEVERITY_LOW: "active_low",
+    Severity.SEVERITY_NONE: "active_none",
+    Severity.SEVERITY_UNKNOWN: "active_unknown",
+}
+
+EVALUATION_RESULT_MAPPING = {
+    License_Policy_Evaluation_Result.RESULT_ALLOWED: "allowed",
+    License_Policy_Evaluation_Result.RESULT_FORBIDDEN: "forbidden",
+    License_Policy_Evaluation_Result.RESULT_IGNORED: "ignored",
+    License_Policy_Evaluation_Result.RESULT_REVIEW_REQUIRED: "review_required",
+    License_Policy_Evaluation_Result.RESULT_UNKNOWN: "unknown",
+}
 
 
 def get_product_by_id(
-    product_id: int, is_product_group: bool = None, with_annotations: bool = False
+    product_id: int,
+    is_product_group: bool = None,
+    with_observation_annotations: bool = False,
+    with_metrics_annotations: bool = False,
 ) -> Optional[Product]:
     try:
         if is_product_group is None:
-            return _add_annotations(Product.objects.all(), False, False).get(id=product_id)
-        return _add_annotations(Product.objects.all(), is_product_group, with_annotations).get(
-            id=product_id, is_product_group=is_product_group
-        )
+            return _add_annotations(Product.objects.all(), False, False, False).get(id=product_id)
+        return _add_annotations(
+            Product.objects.all(), is_product_group, with_observation_annotations, with_metrics_annotations
+        ).get(id=product_id, is_product_group=is_product_group)
     except Product.DoesNotExist:
         return None
 
@@ -39,7 +61,9 @@ def get_product_by_name(name: str, is_product_group: bool = None) -> Optional[Pr
         return None
 
 
-def get_products(is_product_group: bool = None, with_annotations: bool = False) -> QuerySet[Product]:
+def get_products(
+    is_product_group: bool = None, with_observation_annotations: bool = False, with_metrics_annotations: bool = False
+) -> QuerySet[Product]:
     user = get_current_user()
 
     if user is None:
@@ -48,7 +72,7 @@ def get_products(is_product_group: bool = None, with_annotations: bool = False)
     products = Product.objects.all()
 
     if is_product_group is not None:
-        products = _add_annotations(products, is_product_group, with_annotations)
+        products = _add_annotations(products, is_product_group, with_observation_annotations, with_metrics_annotations)
 
     if not user.is_superuser:
         product_members = Product_Member.objects.filter(product=OuterRef("pk"), user=user)
@@ -83,12 +107,18 @@ def get_products(is_product_group: bool = None, with_annotations: bool = False)
     return products
 
 
-def _add_annotations(queryset: QuerySet, is_product_group: bool, with_annotations: bool) -> QuerySet:
-    if not with_annotations:
+def _add_annotations(
+    queryset: QuerySet, is_product_group: bool, with_observation_annotations: bool, with_metrics_annotations: bool
+) -> QuerySet:
+    if not with_observation_annotations and not with_metrics_annotations:
         return queryset
 
-    queryset = _add_observation_annotations(queryset, is_product_group)
-    queryset = _add_license_annotations(queryset, is_product_group)
+    if with_observation_annotations:
+        queryset = _add_observation_annotations(queryset, is_product_group)
+        queryset = _add_license_annotations(queryset, is_product_group)
+    elif with_metrics_annotations:
+        queryset = _add_observation_metrics_annotations(queryset, is_product_group)
+        queryset = _add_license_metrics_annotations(queryset, is_product_group)
     return queryset
 
 
@@ -136,6 +166,50 @@ def _add_observation_annotations(queryset: QuerySet, is_product_group: bool) ->
     return queryset
 
 
+def _add_observation_metrics_annotations(queryset: QuerySet, is_product_group: bool) -> QuerySet:
+    subquery_active_critical = (
+        _get_product_group_metrics_subquery(Severity.SEVERITY_CRITICAL)
+        if is_product_group
+        else _get_product_metrics_subquery(Severity.SEVERITY_CRITICAL)
+    )
+    subquery_active_high = (
+        _get_product_group_metrics_subquery(Severity.SEVERITY_HIGH)
+        if is_product_group
+        else _get_product_metrics_subquery(Severity.SEVERITY_HIGH)
+    )
+    subquery_active_medium = (
+        _get_product_group_metrics_subquery(Severity.SEVERITY_MEDIUM)
+        if is_product_group
+        else _get_product_metrics_subquery(Severity.SEVERITY_MEDIUM)
+    )
+    subquery_active_low = (
+        _get_product_group_metrics_subquery(Severity.SEVERITY_LOW)
+        if is_product_group
+        else _get_product_metrics_subquery(Severity.SEVERITY_LOW)
+    )
+    subquery_active_none = (
+        _get_product_group_metrics_subquery(Severity.SEVERITY_NONE)
+        if is_product_group
+        else _get_product_metrics_subquery(Severity.SEVERITY_NONE)
+    )
+    subquery_active_unknown = (
+        _get_product_group_metrics_subquery(Severity.SEVERITY_UNKNOWN)
+        if is_product_group
+        else _get_product_metrics_subquery(Severity.SEVERITY_UNKNOWN)
+    )
+
+    queryset = queryset.annotate(
+        active_critical_observation_count=Coalesce(subquery_active_critical, 0),
+        active_high_observation_count=Coalesce(subquery_active_high, 0),
+        active_medium_observation_count=Coalesce(subquery_active_medium, 0),
+        active_low_observation_count=Coalesce(subquery_active_low, 0),
+        active_none_observation_count=Coalesce(subquery_active_none, 0),
+        active_unknown_observation_count=Coalesce(subquery_active_unknown, 0),
+    )
+
+    return queryset
+
+
 def _add_license_annotations(queryset: QuerySet, is_product_group: bool) -> QuerySet:
     settings = Settings.load()
     if settings.feature_license_management:
@@ -176,6 +250,46 @@ def _add_license_annotations(queryset: QuerySet, is_product_group: bool) -> Quer
     return queryset
 
 
+def _add_license_metrics_annotations(queryset: QuerySet, is_product_group: bool) -> QuerySet:
+    settings = Settings.load()
+    if settings.feature_license_management:
+        subquery_license_forbidden = (
+            _get_product_group_license_metrics_subquery(License_Policy_Evaluation_Result.RESULT_FORBIDDEN)
+            if is_product_group
+            else _get_product_license_metrics_subquery(License_Policy_Evaluation_Result.RESULT_FORBIDDEN)
+        )
+        subquery_license_review_required = (
+            _get_product_group_license_metrics_subquery(License_Policy_Evaluation_Result.RESULT_REVIEW_REQUIRED)
+            if is_product_group
+            else _get_product_license_metrics_subquery(License_Policy_Evaluation_Result.RESULT_REVIEW_REQUIRED)
+        )
+        subquery_license_unknown = (
+            _get_product_group_license_metrics_subquery(License_Policy_Evaluation_Result.RESULT_UNKNOWN)
+            if is_product_group
+            else _get_product_license_metrics_subquery(License_Policy_Evaluation_Result.RESULT_UNKNOWN)
+        )
+        subquery_license_allowed = (
+            _get_product_group_license_metrics_subquery(License_Policy_Evaluation_Result.RESULT_ALLOWED)
+            if is_product_group
+            else _get_product_license_metrics_subquery(License_Policy_Evaluation_Result.RESULT_ALLOWED)
+        )
+        subquery_license_ignored = (
+            _get_product_group_license_metrics_subquery(License_Policy_Evaluation_Result.RESULT_IGNORED)
+            if is_product_group
+            else _get_product_license_metrics_subquery(License_Policy_Evaluation_Result.RESULT_IGNORED)
+        )
+
+        queryset = queryset.annotate(
+            forbidden_licenses_count=Coalesce(subquery_license_forbidden, 0),
+            review_required_licenses_count=Coalesce(subquery_license_review_required, 0),
+            unknown_licenses_count=Coalesce(subquery_license_unknown, 0),
+            allowed_licenses_count=Coalesce(subquery_license_allowed, 0),
+            ignored_licenses_count=Coalesce(subquery_license_ignored, 0),
+        )
+
+    return queryset
+
+
 def _get_product_observation_subquery(severity: str) -> Subquery:
     branch_filter = Q(branch__is_default_branch=True) | (
         Q(branch__isnull=True) & Q(product__repository_default_branch__isnull=True)
@@ -216,6 +330,23 @@ def _get_product_group_observation_subquery(severity: str) -> Subquery:
     )
 
 
+def _get_product_metrics_subquery(severity: str) -> Subquery:
+    return Subquery(
+        Product_Metrics.objects.filter(product=OuterRef("pk"), date=date.today()).values(SEVERITY_MAPPING[severity]),
+        output_field=IntegerField(),
+    )
+
+
+def _get_product_group_metrics_subquery(severity: str) -> Subquery:
+    return Subquery(
+        Product_Metrics.objects.filter(product__product_group=OuterRef("pk"), date=date.today())
+        .values("product__product_group")
+        .annotate(total=Sum(SEVERITY_MAPPING[severity]))
+        .values("total"),
+        output_field=IntegerField(),
+    )
+
+
 def _get_product_license_subquery(evaluation_result: str) -> Subquery:
     branch_filter = Q(branch__is_default_branch=True) | (
         Q(branch__isnull=True) & Q(product__repository_default_branch__isnull=True)
@@ -252,3 +383,22 @@ def _get_product_group_license_subquery(evaluation_result: str) -> Subquery:
         .values("count"),
         output_field=IntegerField(),
     )
+
+
+def _get_product_license_metrics_subquery(evaluation_result: str) -> Subquery:
+    return Subquery(
+        Product_License_Metrics.objects.filter(product=OuterRef("pk"), date=date.today()).values(
+            EVALUATION_RESULT_MAPPING[evaluation_result]
+        ),
+        output_field=IntegerField(),
+    )
+
+
+def _get_product_group_license_metrics_subquery(evaluation_result: str) -> Subquery:
+    return Subquery(
+        Product_License_Metrics.objects.filter(product__product_group=OuterRef("pk"), date=date.today())
+        .values("product__product_group")
+        .annotate(total=Sum(EVALUATION_RESULT_MAPPING[evaluation_result]))
+        .values("total"),
+        output_field=IntegerField(),
+    )

backend/application/core/services/security_gate.py

@@ -93,7 +93,9 @@ def _calculate_active_product_security_gate(product: Product) -> bool:
             product.security_gate_threshold_unknown if product.security_gate_threshold_unknown else 0
         )
 
-    annotated_product = get_product_by_id(product_id=product.pk, is_product_group=False, with_annotations=True)
+    annotated_product = get_product_by_id(
+        product_id=product.pk, is_product_group=False, with_observation_annotations=True
+    )
     if not annotated_product:
         raise ValueError(f"Product {product.pk} not found while calculating security gate.")
 
@@ -123,7 +125,9 @@ def _calculate_active_product_security_gate(product: Product) -> bool:
 def _calculate_active_config_security_gate(product: Product) -> bool:
     settings = Settings.load()
 
-    annotated_product = get_product_by_id(product_id=product.pk, is_product_group=False, with_annotations=True)
+    annotated_product = get_product_by_id(
+        product_id=product.pk, is_product_group=False, with_observation_annotations=True
+    )
     if not annotated_product:
         raise ValueError(f"Product {product.pk} not found while calculating security gate.")
 

backend/application/import_observations/services/import_observations.py

@@ -536,6 +536,9 @@ def process_license_components(  # pylint: disable=too-many-statements disable=t
         vulnerability_check.last_import_licenses_updated = len(components_updated)
         vulnerability_check.last_import_licenses_deleted = components_deleted
 
+        vulnerability_check.product.last_license_change = timezone.now()
+        vulnerability_check.product.save()
+
     if scanner:
         vulnerability_check.scanner = scanner
     vulnerability_check.save()