Source code for swh.vulns.storage.interface

# Copyright (C) 2026  The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information

from datetime import datetime
from typing import Any, Dict, List, Optional, Protocol, runtime_checkable

from swh.core.api import remote_api_endpoint
from swh.model.swhids import CoreSWHID
from swh.vulns.osv.model import OSVVulnerabilityEvent




@runtime_checkable
class VulnerabilitiesStorageInterface(Protocol):
    """Storage interface to add and get vulnerability data"""



    @remote_api_endpoint("osv_vulnerability_report/add")
    def osv_vulnerability_report_add(
        self,
        vulnerability_id: str,
        processing_date: datetime,
        json_report: Dict[str, Any],
    ):
        """Add last processing date for an OSV vulnerability report
        and the full JSON report."""
        ...




    @remote_api_endpoint("osv_vulnerability_report/last_processing_date")
    def osv_vulnerability_report_last_processing_date(
        self, vulnerability_id: str
    ) -> Optional[datetime]:
        """Get the last processing data for an OSV vulnerability report."""
        ...




    @remote_api_endpoint("osv_vulnerability_report/get")
    def osv_vulnerability_report_get(
        self,
        vulnerability_id: str,
    ) -> Optional[Dict[str, Any]]:
        """Get full JSON report of an OSV vulnerability event."""
        ...




    @remote_api_endpoint("osv_vulnerability_event/add")
    def osv_vulnerability_event_add(
        self, vulnerabilities_events: List[OSVVulnerabilityEvent]
    ):
        """Add OSV vulnerability events in storage."""
        ...




    @remote_api_endpoint("osv_vulnerability_event/get_by_id")
    def osv_vulnerability_event_get_by_id(
        self, vulnerabilities_ids: List[str]
    ) -> List[OSVVulnerabilityEvent]:
        """Get vulnerability events for a given OSV vulnerability."""
        ...




    @remote_api_endpoint("osv_vulnerability_event/get_by_swhid")
    def osv_vulnerability_event_get_by_swhid(
        self, swhids: List[CoreSWHID]
    ) -> List[OSVVulnerabilityEvent]:
        """Get OSV vulnerability events related to a list of SWHIDs."""
        ...




    @remote_api_endpoint("osv_vulnerability_event/get_by_origin_url")
    def osv_vulnerability_event_get_by_origin_url(
        self, origin_urls: List[str]
    ) -> List[OSVVulnerabilityEvent]:
        """Get OSV vulnerability events related to a list of origin URLs."""
        ...