Merge pull request #2112 from aboutcode-org/onetime_pipeline

Support running pipeline only once

Author: keshav-space

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>`_

docs/source/tutorial_add_importer_pipeline.rst

@@ -152,7 +152,7 @@ At this point, an example importer will look like this:
 .. code-block:: python
     :caption: vulnerabilities/pipelines/example_importer.py
     :linenos:
-    :emphasize-lines: 16-17, 20-21, 23-24
+    :emphasize-lines: 17-18, 21-22, 24-25
 
     from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipeline
 
@@ -165,6 +165,7 @@ At this point, an example importer will look like this:
         license_url = "https://exmaple.org/license/"
         spdx_license_expression = "CC-BY-4.0"
         importer_name = "Example Importer"
+        run_once = False
 
         @classmethod
         def steps(cls):
@@ -196,7 +197,7 @@ version management from `univers <https://github.com/aboutcode-org/univers>`_.
 .. code-block:: python
     :caption: vulnerabilities/pipelines/example_importer.py
     :linenos:
-    :emphasize-lines: 34-35, 37-40
+    :emphasize-lines: 35-36, 38-41
 
     from datetime import datetime
     from datetime import timezone
@@ -223,6 +224,7 @@ version management from `univers <https://github.com/aboutcode-org/univers>`_.
         license_url = "https://example.org/license/"
         spdx_license_expression = "CC-BY-4.0"
         importer_name = "Example Importer"
+        run_once = False
 
         @classmethod
         def steps(cls):
@@ -303,6 +305,17 @@ version management from `univers <https://github.com/aboutcode-org/univers>`_.
    Implement ``on_failure`` to handle cleanup in case of pipeline failure.
    Cleanup of downloaded archives or cloned repos is necessary to avoid potential resource leakage.
 
+.. tip::
+
+   Set ``run_once`` to ``True`` if pipeline is meant to be run once.
+
+   - 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 ``is_active` field to ``True`` via a migration.
+
+
 .. note::
 
    | Use ``make valid`` to format your new code using black and isort automatically.

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

vulnerabilities/templates/pipeline_dashboard.html

@@ -80,7 +80,11 @@ <h1>Pipeline Dashboard</h1>
                                     <div class="column is-one-quarter">{{ schedule.pipeline_id }}</div>
                                     <div class="column is-one-eighth has-text-grey">{{ schedule.is_active|yesno:"Yes,No" }}</div>
                                     <div class="column is-one-eighth has-text-grey">
-                                        {{ schedule.run_interval }} hour{{ schedule.run_interval|pluralize }}
+                                        {% if schedule.is_run_once %}
+                                            Once
+                                        {% else %}
+                                            {{ schedule.run_interval }} hour{{ schedule.run_interval|pluralize }}
+                                        {% endif %}
                                     </div>
                                     <div class="column is-one-eighth">
                                         <span class="is-flex is-align-items-center">
@@ -95,7 +99,11 @@ <h1>Pipeline Dashboard</h1>
                                         {% endif %}
                                     </div>
                                     <div class="column is-one-fifth has-text-grey">
-                                        {{ schedule.next_run_date|date:"Y-m-d" }}
+                                        {% if schedule.next_run_date %}
+                                            {{ schedule.next_run_date|date:"Y-m-d" }}
+                                        {% else %}
+                                            N/A
+                                        {% endif %}
                                     </div>
                                 </div>
                             </a>

vulnerabilities/tests/test_tasks.py

@@ -0,0 +1,98 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+
+from unittest import mock
+
+from django.test import TestCase
+
+from vulnerabilities.models import PipelineRun
+from vulnerabilities.models import PipelineSchedule
+from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
+from vulnerabilities.tasks import execute_pipeline
+
+
+class OneTimePipeline(VulnerableCodeBaseImporterPipelineV2):
+    pipeline_id = "one_time_pipeline_test"
+    run_once = True
+
+    def collect_advisories(self):
+        return []
+
+    def advisories_count(self):
+        return 0
+
+
+class NotOneTimePipeline(VulnerableCodeBaseImporterPipelineV2):
+    pipeline_id = "not_one_time_pipeline_test"
+    run_once = False
+
+    def collect_advisories(self):
+        return []
+
+    def advisories_count(self):
+        return 0
+
+
+class TestOneTimePipelineExecution(TestCase):
+    @mock.patch("vulnerabilities.models.PipelineSchedule.create_new_job")
+    @mock.patch(
+        "vulnerabilities.models.PipelineSchedule.pipeline_class",
+        new_callable=mock.PropertyMock,
+    )
+    def test_onetime_pipeline_deactivation(self, mock_pipeline_class, mock_create_job):
+        mock_create_job.return_value = True
+        mock_pipeline_class.return_value = OneTimePipeline
+
+        ps, _ = PipelineSchedule.objects.get_or_create(
+            pipeline_id=OneTimePipeline.pipeline_id,
+            defaults={
+                "is_run_once": OneTimePipeline.run_once,
+            },
+        )
+
+        self.assertTrue(ps.is_run_once)
+        self.assertTrue(ps.is_active)
+
+        run = PipelineRun.objects.create(
+            pipeline=ps,
+        )
+        execute_pipeline(ps.pipeline_id, run.run_id)
+
+        ps.refresh_from_db()
+        self.assertTrue(ps.is_run_once)
+        self.assertFalse(ps.is_active)
+
+    @mock.patch("vulnerabilities.models.PipelineSchedule.create_new_job")
+    @mock.patch(
+        "vulnerabilities.models.PipelineSchedule.pipeline_class",
+        new_callable=mock.PropertyMock,
+    )
+    def test_normal_pipeline_no_deactivation(self, mock_pipeline_class, mock_create_job):
+        mock_create_job.return_value = True
+        mock_pipeline_class.return_value = NotOneTimePipeline
+
+        ps, _ = PipelineSchedule.objects.get_or_create(
+            pipeline_id=NotOneTimePipeline.pipeline_id,
+            defaults={
+                "is_run_once": NotOneTimePipeline.run_once,
+            },
+        )
+
+        self.assertFalse(ps.is_run_once)
+        self.assertTrue(ps.is_active)
+
+        run = PipelineRun.objects.create(
+            pipeline=ps,
+        )
+        execute_pipeline(ps.pipeline_id, run.run_id)
+
+        ps.refresh_from_db()
+        self.assertFalse(ps.is_run_once)
+        self.assertTrue(ps.is_active)