Class ImportRunner

vulnerabilities/import_runner.py:40–149

The ImportRunner is responsible for inserting and updating data about vulnerabilities and affected/unaffected/fixed packages in the database. The main goal for the implementation is correctness Correctness: - There must be no duplicates in the database (should be enforced b

class ImportRunner:
    """
    The ImportRunner is responsible for inserting and updating data about vulnerabilities and
    affected/unaffected/fixed packages in the database. The main goal for the implementation
    is correctness

    Correctness:
        - There must be no duplicates in the database (should be enforced by the schema).
        - No valid data from the data source must be skipped or truncated.
    """

    def __init__(self, importer_class: Importer):
        self.importer_class = importer_class

    def run(self) -> None:
        """
        Create a data source for the given importer and store the data retrieved in the database.
        """
        importer_name = self.importer_class.qualified_name
        importer_class = self.importer_class
        logger.info(f"Starting import for {importer_name}")
        advisory_datas = importer_class().advisory_data()
        count = self.process_advisories(advisory_datas=advisory_datas, importer_name=importer_name)
        logger.info(f"Finished import for {importer_name}. Imported {count} advisories.")

    def do_import(self, advisories) -> None:
        advisory_importer = DefaultImporter(advisories=advisories)
        logger.info(f"Running importer: {advisory_importer.qualified_name}")
        importer_name = advisory_importer.qualified_name
        advisories = []
        for advisory in advisory_importer.interesting_advisories:
            if advisory.date_imported:
                continue
            logger.info(f"Processing advisory: {advisory!r}")
            advisory_data = None
            inferences = None
            try:
                advisory_data = advisory.to_advisory_data()
                inferences = advisory_importer.get_inferences(advisory_data=advisory_data)
                process_inferences(
                    inferences=inferences,
                    advisory=advisory,
                    improver_name=importer_name,
                )
            except Exception:
                from pprint import pformat

                logger.warning(
                    f"Failed to process advisory:\n{pformat(advisory_data.to_dict())}\n\n"
                    f"with error:\n{traceback_format_exc()}\n\n"
                )
        logger.info("Finished importing using %s.", advisory_importer.__class__.qualified_name)

    def process_advisories(
        self, advisory_datas: Iterable[AdvisoryData], importer_name: str
    ) -> List:
        """
        Insert advisories into the database