Support running pipeline only once

docs/source/contributing.rst

@@ -17,7 +17,7 @@ resources to help you get started.
 Do Your Homework
 ----------------
 
-Before adding a contribution or create a new issue, take a look at the project’s
+Before adding a contribution or create a new issue, take a look at the project's
 `README <https://github.com/aboutcode-org/vulnerablecode>`_, read through our
 `documentation <https://vulnerablecode.readthedocs.io/en/latest/>`_,
 and browse existing `issues <https://github.com/aboutcode-org/vulnerablecode/issues>`_,
@@ -73,7 +73,7 @@ overlooked. We value any suggestions to improve
 
 .. tip::
     Our documentation is treated like code. Make sure to check our
-    `writing guidelines <https://scancode-toolkit.readthedocs.io/en/latest/contribute/contrib_doc.html>`_
+    `writing guidelines <https://scancode-toolkit.readthedocs.io/en/stable/contribute/contrib_doc.html>`_
     to help guide new users.
 
 Other Ways
@@ -87,7 +87,7 @@ questions, and interact with us and other community members on
 Helpful Resources
 -----------------
 
-- Review our `comprehensive guide <https://scancode-toolkit.readthedocs.io/en/latest/contribute/index.html>`_
+- Review our `comprehensive guide <https://scancode-toolkit.readthedocs.io/en/stable/contribute/index.html>`_
   for more details on how to add quality contributions to our codebase and documentation
 - Check this free resource on `How to contribute to an open source project on github <https://egghead.io/lessons/javascript-identifying-how-to-contribute-to-an-open-source-project-on-github>`_
 - Follow `this wiki page <https://aboutcode.readthedocs.io/en/latest/contributing/writing_good_commit_messages.html>`_

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/migrations/0110_pipelineschedule_is_run_once.py

@@ -0,0 +1,22 @@
+# Generated by Django 4.2.22 on 2026-01-08 13:41
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("vulnerabilities", "0109_alter_advisoryseverity_scoring_elements_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="pipelineschedule",
+            name="is_run_once",
+            field=models.BooleanField(
+                db_index=True,
+                default=False,
+                help_text="When set to True, this Pipeline will run only once.",
+            ),
+        ),
+    ]

vulnerabilities/models.py

@@ -2273,6 +2273,13 @@ class PipelineSchedule(models.Model):
         ),
     )
 
+    is_run_once = models.BooleanField(
+        null=False,
+        db_index=True,
+        default=False,
+        help_text=("When set to True, this Pipeline will run only once."),
+    )
+
     live_logging = models.BooleanField(
         null=False,
         db_index=True,

vulnerabilities/pipelines/__init__.py

@@ -169,6 +169,10 @@ class VulnerableCodeBaseImporterPipeline(VulnerableCodePipeline):
     importer_name = None
     advisory_confidence = MAX_CONFIDENCE
 
+    # When set to true pipeline is run only once.
+    # To rerun onetime pipeline reset is_active field to True via migration.
+    run_once = False
+
     @classmethod
     def steps(cls):
         return (
@@ -262,6 +266,10 @@ class VulnerableCodeBaseImporterPipelineV2(VulnerableCodePipeline):
     repo_url = None
     ignorable_versions = []
 
+    # When set to true pipeline is run only once.
+    # To rerun onetime pipeline reset is_active field to True via migration.
+    run_once = False
+
     @classmethod
     def steps(cls):
         return (

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")