swh.osv.commit_ranges module#

swh.osv.commit_ranges.init_db(conn: Connection) → None[source]#

class swh.osv.commit_ranges.DocProcessor(stub: swh.graph.grpc.swhgraph_pb2_grpc.TraversalServiceStub, session: aiohttp.client.ClientSession, conn: sqlite3.Connection, origin_denormalizer: swh.osv.map_origins.OriginDenormalizer, root_revisions_cache: swh.osv.utils.Cache[dict[tuple[str, ...], list[str]]])[source]#

Bases: object

stub: TraversalServiceStub#

session: ClientSession#

conn: Connection#

origin_denormalizer: OriginDenormalizer#

root_revisions_cache: Cache[dict[tuple[str, ...], list[str]]]#

total_invalid_commit_ids = 0#

total_unknown_commits = 0#

total_skipped_due_to_invalid_commit = 0#

total_skipped_due_to_unknown_commit = 0#

total_skipped_due_to_unimplemented_event_types = 0#

total_vulnerable = 0#

total_successes = 0#

add_unknown_swhid(swhid) → None[source]#

get_unknown_swhids(swhids: Iterable[str]) → set[str][source]#: Given an iterable of SWHIDs, returns which ones are known to be unknown

async process_docs(arg)[source]#

async expand_git_range(range_: dict) → dict[str, set[str]] | None[source]#

async process_git_range(commits_by_event_type: dict[str, set[str]]) → None[source]#

async swh.osv.commit_ranges.main(conn: Connection) → None[source]#