Function insert_advisory_v2

vulnerabilities/pipes/advisory.py:292–419

(
    advisory: AdvisoryDataV2,
    pipeline_id: str,
    logger: Callable,
    datasource_id: str,
    precedence: int = 0,
)

@transaction.atomic
def insert_advisory_v2(
    advisory: AdvisoryDataV2,
    pipeline_id: str,
    logger: Callable,
    datasource_id: str,
    precedence: int = 0,
):
    from vulnerabilities.models import ImpactedPackage
    from vulnerabilities.models import PackageV2
    from vulnerabilities.utils import compute_content_id_v2

    advisory_obj = None
    created = False
    content_id = compute_content_id_v2(advisory_data=advisory)
    try:
        default_data = {
            "datasource_id": datasource_id,
            "pipeline_id": pipeline_id,
            "advisory_id": advisory.advisory_id,
            "avid": f"{datasource_id}/{advisory.advisory_id}",
            "summary": advisory.summary,
            "date_published": advisory.date_published,
            "original_advisory_text": advisory.original_advisory_text,
            "url": advisory.url,
            "precedence": precedence,
        }

        advisory_obj, created = AdvisoryV2.objects.get_or_create(
            advisory_id=advisory.advisory_id,
            datasource_id=datasource_id,
            pipeline_id=pipeline_id,
            unique_content_id=content_id,
            defaults=default_data,
        )
    except AdvisoryV2.MultipleObjectsReturned:
        logger(
            f"Multiple Advisories returned: unique_content_id: {content_id}, url: {advisory.url}, advisory: {advisory!r}"
        )
        raise
    except Exception as e:
        logger(
            f"Error while processing {advisory!r} with aliases {advisory.aliases!r}: {e!r} \n {traceback_format_exc()}",
            level=logging.ERROR,
        )
        raise

    if not created:
        return advisory_obj

    AdvisoryV2.objects.filter(
        avid=f"{datasource_id}/{advisory.advisory_id}",
        is_latest=True,
    ).update(is_latest=False)
    advisory_obj.is_latest = True
    advisory_obj.save()

    aliases = get_or_create_advisory_aliases(aliases=advisory.aliases)
    references = get_or_create_advisory_references(references=advisory.references)