swh.vulns.storage.postgresql module#

class swh.vulns.storage.postgresql.VulnerabilitiesStorage(db, min_pool_conns: int = 1, max_pool_conns: int = 10)[source]#

Bases: VulnerabilitiesStorageInterface

Parameters:: db – either a libpq connection string, or a psycopg connection

current_version = 1#

get_db()[source]#

put_db(db)[source]#

osv_vulnerability_report_add(vulnerability_id: str, processing_date: datetime, json_report: Dict[str, Any])[source]#: Add last processing date for an OSV vulnerability report and the full JSON report.

osv_vulnerability_report_last_processing_date(vulnerability_id: str) → datetime | None[source]#: Get the last processing data for an OSV vulnerability report.

osv_vulnerability_report_get(vulnerability_id: str) → Dict[str, Any] | None[source]#: Get full JSON report of an OSV vulnerability event.

vulnerability_event_cols = ['vulnerability_id', 'vulnerability_severity', 'event_type', 'origin_url', 'swhid', 'version']#

osv_vulnerability_event_add(vulnerabilities_events: List[OSVVulnerabilityEvent])[source]#: Add OSV vulnerability events in storage.

osv_vulnerability_event_get_by_id(vulnerabilities_ids: List[str]) → List[OSVVulnerabilityEvent][source]#: Get vulnerability events for a given OSV vulnerability.

osv_vulnerability_event_get_by_swhid(swhids: List[CoreSWHID]) → List[OSVVulnerabilityEvent][source]#: Get OSV vulnerability events related to a list of SWHIDs.

osv_vulnerability_event_get_by_origin_url(origin_urls: List[str]) → List[OSVVulnerabilityEvent][source]#: Get OSV vulnerability events related to a list of origin URLs.