Function get_affected_advisories_bulk

vulnerabilities/api_v3.py:566–888

(packages, max_advisories, base_url, reachability=False)

def get_affected_advisories_bulk(packages, max_advisories, base_url, reachability=False):
    package_ids = []
    package_ids_with_multiple_importers = []
    packages_without_multiple_importers = []

    for p in packages:
        package_ids.append(p.id)

        if p.type in TYPES_WITH_MULTIPLE_IMPORTERS:
            package_ids_with_multiple_importers.append(p.id)
        else:
            packages_without_multiple_importers.append(p)

    result = {}

    impacts = (
        ImpactedPackageAffecting.objects.filter(
            package_id__in=package_ids,
            impacted_package__advisory__is_latest=True,
            impacted_package__advisory___all_impacts_unfurled_at__isnull=False,
        )
        .values(
            "package_id",
            "impacted_package__advisory_id",
        )
        .annotate(
            fixed_by_packages=ArrayAgg(
                "impacted_package__fixed_by_packages__package_url",
                distinct=True,
                filter=Q(impacted_package__fixed_by_packages__package_url__isnull=False),
            )
        )
    )

    impact_by_package_and_advisory = {
        (
            row["package_id"],
            row["impacted_package__advisory_id"],
        ): row["fixed_by_packages"]
        or []
        for row in impacts
    }

    # Package types with multiple importers

    advisory_sets = list(
        AdvisorySet.objects.filter(
            package_id__in=package_ids_with_multiple_importers,
            relation_type="affecting",
        )
        .select_related("primary_advisory")
        .prefetch_related(
            Prefetch(
                "aliases",
                queryset=AdvisoryAlias.objects.only("alias"),
            ),
        )
        .annotate(