Add queryset method to bulk-create packages from list of PURLs

Signed-off-by: Keshav Priyadarshi <[email protected]>

Author: keshav-space

vulnerabilities/models.py

@@ -16,6 +16,8 @@
 from functools import cached_property
 from itertools import groupby
 from operator import attrgetter
+from traceback import format_exc as traceback_format_exc
+from typing import List
 from typing import Union
 from urllib.parse import urljoin
 
@@ -3067,6 +3069,41 @@ def get_or_create_from_purl(self, purl: Union[PackageURL, str]):
 
         return package, is_created
 
+    def bulk_get_or_create_from_purls(self, purls: List[Union[PackageURL, str]]):
+        """
+        Return new or existing Packages given ``purls`` list of PackageURL object or PURL string.
+        """
+        purl_strings = [str(p) for p in purls]
+        existing_packages = PackageV2.objects.filter(package_url__in=purl_strings)
+        existing_purls = set(existing_packages.values_list("package_url", flat=True))
+
+        all_packages = list(existing_packages)
+        packages_to_create = []
+        for purl in purls:
+            if str(purl) in existing_purls:
+                continue
+
+            purl_dict = purl_to_dict(purl)
+            purl = PackageURL(**purl_dict)
+
+            normalized = normalize_purl(purl=purl)
+            for name, value in purl_to_dict(normalized).items():
+                setattr(self, name, value)
+
+            purl_dict["package_url"] = str(normalized)
+            purl_dict["plain_package_url"] = str(utils.plain_purl(normalized))
+
+            packages_to_create.append(PackageV2(**purl_dict))
+
+        try:
+            new_packages = PackageV2.objects.bulk_create(packages_to_create)
+        except Exception as e:
+            logging.error(f"Error creating PackageV2: {e} \n {traceback_format_exc()}")
+            return []
+
+        all_packages.extend(new_packages)
+        return all_packages
+
     def only_vulnerable(self):
         return self._vulnerable(True)