Migrate PyPa importer to aboutcode pipeline

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

Author: keshav-space

vulnerabilities/pipelines/pypa_importer.py

@@ -7,60 +7,64 @@
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
 import logging
-import os
 from pathlib import Path
 from typing import Iterable
 
 import saneyaml
+from fetchcode.vcs import fetch_via_vcs
 
 from vulnerabilities.importer import AdvisoryData
-from vulnerabilities.importer import Importer
 from vulnerabilities.importers.osv import parse_advisory_data
+from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipeline
 from vulnerabilities.utils import get_advisory_url
 
-logger = logging.getLogger(__name__)
+module_logger = logging.getLogger(__name__)
 
 
-class PyPaImporter(Importer):
-    license_url = "https://github.com/pypa/advisory-database/blob/main/LICENSE"
+class PyPaImporterPipeline(VulnerableCodeBaseImporterPipeline):
+    """Collect advisories from PyPA GitHub repository."""
+
     spdx_license_expression = "CC-BY-4.0"
+    license_url = "https://github.com/pypa/advisory-database/blob/main/LICENSE"
     repo_url = "git+https://github.com/pypa/advisory-database"
     importer_name = "Pypa Importer"
 
-    def advisory_data(self) -> Iterable[AdvisoryData]:
-        try:
-            vcs_response = self.clone(repo_url=self.repo_url)
-            path = Path(vcs_response.dest_dir)
-            for advisory_url, raw_data in fork_and_get_files(base_path=path):
-                yield parse_advisory_data(
-                    raw_data=raw_data,
-                    supported_ecosystems=["pypi"],
-                    advisory_url=advisory_url,
-                )
-        finally:
-            if self.vcs_response:
-                self.vcs_response.delete()
+    @classmethod
+    def steps(cls):
+        return (
+            cls.clone,
+            cls.collect_and_store_advisories,
+            cls.import_new_advisories,
+            cls.clean_downloads,
+        )
 
+    def clone(self):
+        self.log(f"Cloning `{self.repo_url}`")
+        self.vcs_response = fetch_via_vcs(self.repo_url)
 
-class ForkError(Exception):
-    pass
+    def advisories_count(self):
+        vulns_directory = Path(self.vcs_response.dest_dir) / "vulns"
+        return sum(1 for _ in vulns_directory.rglob("*.yaml"))
 
+    def collect_advisories(self) -> Iterable[AdvisoryData]:
+        base_directory = Path(self.vcs_response.dest_dir)
+        vulns_directory = base_directory / "vulns"
+        self.advisories_count = sum(1 for _ in vulns_directory.rglob("*.yaml"))
 
-def fork_and_get_files(base_path) -> dict:
-    """
-    Yield advisorie data mappings from the PyPA GitHub repository at ``url``.
-    """
-    advisory_dirs = os.path.join(base_path, "vulns")
-    for root, _, files in os.walk(advisory_dirs):
-        for file in files:
-            path = os.path.join(root, file)
-            if not file.endswith(".yaml"):
-                logger.warning(f"Unsupported non-YAML PyPA advisory file: {path}")
-                continue
+        for advisory in vulns_directory.rglob("*.yaml"):
             advisory_url = get_advisory_url(
-                file=Path(path),
-                base_path=base_path,
+                file=advisory,
+                base_path=base_directory,
                 url="https://github.com/pypa/advisory-database/blob/main/",
             )
-            with open(path) as f:
-                yield advisory_url, saneyaml.load(f.read())
+            advisory_dict = saneyaml.load(advisory.read_text())
+            yield parse_advisory_data(
+                raw_data=advisory_dict,
+                supported_ecosystems=["pypi"],
+                advisory_url=advisory_url,
+            )
+
+    def clean_downloads(self):
+        if self.vcs_response:
+            self.log(f"Removing cloned repository")
+            self.vcs_response.delete()