Deactivate onetime pipeline after first run

- To rerun onetime pipeline, reset `is_active` to `True` via a migration,
    pipeline will run one more time and then deactivate.
- To convert a onetime pipeline to a regular pipeline,
    set the `run_once` class variable to `False` and reset
    the `is_active` field to `True` via a migration.

Signed-off-by: Keshav Priyadarshi <git@keshav.space>

Author: keshav-space

vulnerabilities/management/commands/run_scheduler.py

@@ -17,16 +17,21 @@
 
 
 def init_pipeline_scheduled():
-    """Initialize schedule jobs for active PipelineSchedule."""
-    active_pipeline_qs = models.PipelineSchedule.objects.filter(is_active=True).order_by(
-        "created_date"
-    )
-    for pipeline_schedule in active_pipeline_qs:
-        if scheduled_job_exists(pipeline_schedule.schedule_work_id):
-            continue
-        new_id = pipeline_schedule.create_new_job()
-        pipeline_schedule.schedule_work_id = new_id
-        pipeline_schedule.save(update_fields=["schedule_work_id"])
+    """
+    Initialize schedule jobs for active PipelineSchedule.
+        - Create new schedule if there is no schedule for active pipeline
+        - Create new schedule if schedule is corrupted for an active pipeline
+        - Delete schedule for inactive pipeline
+    """
+    pipeline_qs = models.PipelineSchedule.objects.order_by("created_date")
+    for pipeline in pipeline_qs:
+        reset_schedule = pipeline.is_active != bool(pipeline.schedule_work_id)
+        if not scheduled_job_exists(pipeline.schedule_work_id):
+            reset_schedule = True
+
+        if reset_schedule:
+            pipeline.schedule_work_id = pipeline.create_new_job()
+            pipeline.save(update_fields=["schedule_work_id"])
 
 
 class Command(rqscheduler.Command):

vulnerabilities/schedules.py

@@ -89,7 +89,15 @@ def update_pipeline_schedule():
     from vulnerabilities.improvers import IMPROVERS_REGISTRY
     from vulnerabilities.models import PipelineSchedule
 
-    pipeline_ids = [*IMPORTERS_REGISTRY.keys(), *IMPROVERS_REGISTRY.keys()]
-
-    PipelineSchedule.objects.exclude(pipeline_id__in=pipeline_ids).delete()
-    [PipelineSchedule.objects.get_or_create(pipeline_id=id) for id in pipeline_ids]
+    pipelines = IMPORTERS_REGISTRY | IMPROVERS_REGISTRY
+
+    PipelineSchedule.objects.exclude(pipeline_id__in=pipelines.keys()).delete()
+    for id, pipeline_class in pipelines.items():
+        run_once = getattr(pipeline_class, "run_once", False)
+
+        PipelineSchedule.objects.get_or_create(
+            pipeline_id=id,
+            defaults={
+                "is_run_once": run_once,
+            },
+        )

vulnerabilities/tasks.py

@@ -17,7 +17,6 @@
 from vulnerabilities import models
 from vulnerabilities.importer import Importer
 from vulnerabilities.improver import Improver
-from vulnerablecode.settings import VULNERABLECODE_PIPELINE_TIMEOUT
 
 logger = logging.getLogger(__name__)
 
@@ -48,6 +47,13 @@ def execute_pipeline(pipeline_id, run_id):
         exitcode = 1
 
     run.set_run_ended(exitcode=exitcode, output=output)
+
+    # Onetime pipeline are inactive after first execution.
+    pipeline = run.pipeline
+    if pipeline.is_run_once:
+        pipeline.is_active = False
+        pipeline.save()
+
     logger.info("Update Run instance with exitcode, output, and end_date")