Add Live Evaluation API endpoint and PyPa live pipeline importer

docker-compose.yml

@@ -23,9 +23,9 @@ services:
   vulnerablecode:
     build: .
     command: /bin/sh -c "
-        ./manage.py migrate &&
-        ./manage.py collectstatic --no-input --verbosity 0 --clear &&
-        gunicorn vulnerablecode.wsgi:application -u nobody -g nogroup --bind :8000 --timeout 600 --workers 8"
+      ./manage.py migrate &&
+      ./manage.py collectstatic --no-input --verbosity 0 --clear &&
+      gunicorn vulnerablecode.wsgi:application -u nobody -g nogroup --bind :8000 --timeout 600 --workers 8"
     env_file:
       - docker.env
     expose:
@@ -60,6 +60,17 @@ services:
       - db
       - vulnerablecode
 
+  vulnerablecode_rqworker_live:
+    build: .
+    command: wait-for-it web:8000 -- python ./manage.py rqworker live
+    env_file:
+      - docker.env
+    volumes:
+      - /etc/vulnerablecode/:/etc/vulnerablecode/
+    depends_on:
+      - vulnerablecode_redis
+      - db
+      - vulnerablecode
 
   nginx:
     image: nginx
@@ -75,9 +86,7 @@ services:
     depends_on:
       - vulnerablecode
 
-
 volumes:
   db_data:
   static:
   vulnerablecode_redis_data:
-

docker.env

@@ -5,4 +5,6 @@ POSTGRES_PASSWORD=vulnerablecode
 VULNERABLECODE_DB_HOST=db
 VULNERABLECODE_STATIC_ROOT=/var/vulnerablecode/static/
 
-VULNERABLECODE_REDIS_HOST=vulnerablecode_redis
+VULNERABLECODE_REDIS_HOST=vulnerablecode_redis
+
+VULNERABLECODE_ENABLE_LIVE_EVALUATION_API=false

vulnerabilities/api_v2.py

@@ -9,6 +9,7 @@
 
 
 from django.db.models import Prefetch
+from django.urls import reverse
 from django_filters import rest_framework as filters
 from drf_spectacular.utils import OpenApiParameter
 from drf_spectacular.utils import extend_schema
@@ -25,6 +26,7 @@
 from rest_framework.reverse import reverse
 from rest_framework.throttling import AnonRateThrottle
 
+from vulnerabilities.importers import LIVE_IMPORTERS_REGISTRY
 from vulnerabilities.models import AdvisoryReference
 from vulnerabilities.models import AdvisorySeverity
 from vulnerabilities.models import AdvisoryV2
@@ -40,7 +42,9 @@
 from vulnerabilities.models import VulnerabilityReference
 from vulnerabilities.models import VulnerabilitySeverity
 from vulnerabilities.models import Weakness
+from vulnerabilities.tasks import enqueue_ad_hoc_pipeline
 from vulnerabilities.throttling import PermissionBasedUserRateThrottle
+from vulnerablecode.settings import VULNERABLECODE_ENABLE_LIVE_EVALUATION_API
 
 
 class CharInFilter(filters.BaseInFilter, filters.CharFilter):
@@ -1293,3 +1297,137 @@ def lookup(self, request):
         return Response(
             AdvisoryPackageV2Serializer(qs, many=True, context={"request": request}).data
         )
+
+
+class LiveEvaluationSerializer(serializers.Serializer):
+    purl = serializers.CharField(help_text="PackageURL to evaluate")
+
+
+class LiveEvaluationViewSet(viewsets.GenericViewSet):
+    serializer_class = LiveEvaluationSerializer
+    throttle_classes = [AnonRateThrottle, PermissionBasedUserRateThrottle]
+
+    @extend_schema(
+        request=LiveEvaluationSerializer,
+        responses={
+            202: {"description": "Live evaluation enqueued successfully; returns Run IDs"},
+            400: {"description": "Invalid request"},
+            500: {"description": "Internal server error"},
+        },
+    )
+    @action(detail=False, methods=["post"])
+    def evaluate(self, request):
+        if not VULNERABLECODE_ENABLE_LIVE_EVALUATION_API:
+            return Response(
+                {"error": "Live evaluation API is disabled."},
+                status=status.HTTP_403_FORBIDDEN,
+            )
+        serializer = self.get_serializer(data=request.data)
+        if not serializer.is_valid():
+            return Response(
+                serializer.errors,
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        purl_string = serializer.validated_data.get("purl")
+
+        try:
+            purl = PackageURL.from_string(purl_string) if purl_string else None
+            if not purl:
+                return Response({"error": "Invalid PackageURL"}, status=status.HTTP_400_BAD_REQUEST)
+        except Exception as e:
+            return Response(
+                {"error": f"Invalid PackageURL: {str(e)}"}, status=status.HTTP_400_BAD_REQUEST
+            )
+
+        importers = [
+            importer
+            for importer in LIVE_IMPORTERS_REGISTRY.values()
+            if hasattr(importer, "supported_types")
+            and purl.type in getattr(importer, "supported_types", [])
+        ]
+
+        if not importers:
+            return Response(
+                {"error": f"No live importers found for purl type '{purl.type}'"},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        # Enqueue all selected importers together and link runs to a new LivePipelineRun
+        importer_ids = [getattr(imp, "pipeline_id", imp.__name__) for imp in importers]
+        live_run_id, run_ids = enqueue_ad_hoc_pipeline(importer_ids, inputs={"purl": purl})
+        runs = [
+            {"importer": importer_ids[idx], "run_id": str(rid)} for idx, rid in enumerate(run_ids)
+        ]
+
+        request_obj = request
+        status_path = reverse("live-evaluation-status", kwargs={"live_run_id": str(live_run_id)})
+
+        if hasattr(request_obj, "build_absolute_uri"):
+            status_url = request_obj.build_absolute_uri(status_path)
+        else:
+            status_url = status_path
+
+        return Response(
+            {
+                "live_run_id": str(live_run_id),
+                "runs": runs,
+                "status_url": status_url,
+            },
+            status=status.HTTP_202_ACCEPTED,
+        )
+
+    @extend_schema(
+        parameters=[
+            OpenApiParameter(
+                name="live_run_id",
+                description="UUID of the live run to check status for",
+                required=True,
+                type={"type": "string"},
+                location=OpenApiParameter.PATH,
+            )
+        ],
+        responses={200: "LivePipelineRun status and importers status"},
+    )
+    @action(detail=False, methods=["get"], url_path=r"status/(?P<live_run_id>[0-9a-f\-]{36})")
+    def status(self, request, live_run_id=None):
+        if not VULNERABLECODE_ENABLE_LIVE_EVALUATION_API:
+            return Response(
+                {"error": "Live evaluation API is disabled."},
+                status=status.HTTP_403_FORBIDDEN,
+            )
+
+        from vulnerabilities.models import LivePipelineRun
+
+        try:
+            live_run = LivePipelineRun.objects.get(run_id=live_run_id)
+        except LivePipelineRun.DoesNotExist:
+            return Response({"detail": "Live run not found."}, status=status.HTTP_404_NOT_FOUND)
+
+        live_run.update_status()
+
+        # Gather status for each importer run
+        importer_statuses = []
+        for run in live_run.pipelineruns.all():
+            importer_statuses.append(
+                {
+                    "importer": run.pipeline.pipeline_id,
+                    "run_id": str(run.run_id),
+                    "status": run.status,
+                    "run_start_date": run.run_start_date,
+                    "run_end_date": run.run_end_date,
+                    "run_exitcode": run.run_exitcode,
+                    "run_output": run.run_output,
+                }
+            )
+
+        response = {
+            "live_run_id": str(live_run.run_id),
+            "overall_status": live_run.status,
+            "created_date": live_run.created_date,
+            "started_date": getattr(live_run, "started_date", None),
+            "completed_date": live_run.completed_date,
+            "purl": live_run.purl,
+            "importers": importer_statuses,
+        }
+        return Response(response)

vulnerabilities/importers/__init__.py

@@ -56,6 +56,7 @@
 from vulnerabilities.pipelines.v2_importers import oss_fuzz as oss_fuzz_v2
 from vulnerabilities.pipelines.v2_importers import postgresql_importer as postgresql_importer_v2
 from vulnerabilities.pipelines.v2_importers import pypa_importer as pypa_importer_v2
+from vulnerabilities.pipelines.v2_importers import pypa_live_importer as pypa_live_importer_v2
 from vulnerabilities.pipelines.v2_importers import pysec_importer as pysec_importer_v2
 from vulnerabilities.pipelines.v2_importers import redhat_importer as redhat_importer_v2
 from vulnerabilities.pipelines.v2_importers import vulnrichment_importer as vulnrichment_importer_v2
@@ -117,3 +118,9 @@
         oss_fuzz.OSSFuzzImporter,
     ]
 )
+
+LIVE_IMPORTERS_REGISTRY = create_registry(
+    [
+        pypa_live_importer_v2.PyPaLiveImporterPipeline,
+    ]
+)

vulnerabilities/migrations/0102_livepipelinerun_pipelinerun_live_pipeline.py

@@ -0,0 +1,48 @@
+# Generated by Django 4.2.22 on 2025-08-25 18:03
+
+from django.db import migrations, models
+import django.db.models.deletion
+import uuid
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("vulnerabilities", "0101_advisorytodov2_todorelatedadvisoryv2_and_more"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="LivePipelineRun",
+            fields=[
+                (
+                    "run_id",
+                    models.UUIDField(
+                        default=uuid.uuid4,
+                        editable=False,
+                        primary_key=True,
+                        serialize=False,
+                        unique=True,
+                    ),
+                ),
+                ("created_date", models.DateTimeField(auto_now_add=True, db_index=True)),
+                ("completed_date", models.DateTimeField(blank=True, editable=False, null=True)),
+                ("status", models.CharField(default="queued", max_length=20)),
+                ("purl", models.CharField(blank=True, max_length=300, null=True)),
+            ],
+            options={
+                "ordering": ["-created_date"],
+            },
+        ),
+        migrations.AddField(
+            model_name="pipelinerun",
+            name="live_pipeline",
+            field=models.ForeignKey(
+                blank=True,
+                null=True,
+                on_delete=django.db.models.deletion.CASCADE,
+                related_name="pipelineruns",
+                to="vulnerabilities.livepipelinerun",
+            ),
+        ),
+    ]

vulnerabilities/models.py

@@ -1972,6 +1972,55 @@ class CodeFixV2(CodeChangeV2):
     )
 
 
+class LivePipelineRun(models.Model):
+    """Represents a single live evaluation run for all compatible importers."""
+
+    run_id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False, unique=True)
+    created_date = models.DateTimeField(auto_now_add=True, db_index=True)
+    completed_date = models.DateTimeField(blank=True, null=True, editable=False)
+    status = models.CharField(max_length=20, default="queued")
+    purl = models.CharField(max_length=300, blank=True, null=True)
+
+    def is_finished(self):
+        return self.status == "finished"
+
+    def update_status(self):
+        runs = list(self.pipelineruns.all())
+        if not runs:
+            self.status = "queued"
+            self.completed_date = None
+            self.save(update_fields=["status", "completed_date"])
+            return
+
+        # Determine aggregate status
+        if all(r.status == PipelineRun.Status.SUCCESS for r in runs):
+            self.status = "finished"
+        elif any(r.status == PipelineRun.Status.FAILURE for r in runs):
+            self.status = "failed"
+        elif any(r.status == PipelineRun.Status.RUNNING for r in runs):
+            self.status = "running"
+        else:
+            # queued or mixed queued
+            self.status = "queued"
+
+        end_times = [r.run_end_date for r in runs if r.run_end_date]
+        completed = None
+        if self.status in ("finished", "failed") and end_times:
+            completed = max(end_times)
+        self.completed_date = completed
+        self.save(update_fields=["status", "completed_date"])
+
+    @property
+    def started_date(self):
+        """Return earliest run_start_date among child runs, if any."""
+        runs = self.pipelineruns.all()
+        start_times = [r.run_start_date for r in runs if r.run_start_date]
+        return min(start_times) if start_times else None
+
+    class Meta:
+        ordering = ["-created_date"]
+
+
 class PipelineRun(models.Model):
     """The Database representation of a pipeline execution."""
 
@@ -1981,6 +2030,14 @@ class PipelineRun(models.Model):
         on_delete=models.CASCADE,
     )
 
+    live_pipeline = models.ForeignKey(
+        "LivePipelineRun",
+        related_name="pipelineruns",
+        on_delete=models.CASCADE,
+        blank=True,
+        null=True,
+    )
+
     run_id = models.UUIDField(
         primary_key=True,
         default=uuid.uuid4,
@@ -2244,8 +2301,7 @@ def append_to_log(self, message, is_multiline=False):
         message = message.strip()
         if not is_multiline:
             message = message.replace("\n", "").replace("\r", "")
-
-        self.log = self.log + message + "\n"
+        self.log = (self.log or "") + message + "\n"
         self.save(update_fields=["log"])
 
     def dequeue(self):
@@ -2342,12 +2398,15 @@ def save(self, *args, **kwargs):
     def pipeline_class(self):
         """Return the pipeline class."""
         from vulnerabilities.importers import IMPORTERS_REGISTRY
+        from vulnerabilities.importers import LIVE_IMPORTERS_REGISTRY
         from vulnerabilities.improvers import IMPROVERS_REGISTRY
 
         if self.pipeline_id in IMPROVERS_REGISTRY:
             return IMPROVERS_REGISTRY.get(self.pipeline_id)
         if self.pipeline_id in IMPORTERS_REGISTRY:
             return IMPORTERS_REGISTRY.get(self.pipeline_id)
+        if self.pipeline_id in LIVE_IMPORTERS_REGISTRY:
+            return LIVE_IMPORTERS_REGISTRY.get(self.pipeline_id)
 
     @property
     def description(self):