swh.osv.map_git_origins module#

swh.osv.map_git_origins.SWH_API_SEMAPHORE = <asyncio.locks.Semaphore object at 0x7f86e4f4c4d0 [unlocked, value:10]>#: Prevents more than 10 concurrent requests to the SWH API

exception swh.osv.map_git_origins.InvalidCommitId[source]#: Bases: ValueError

async swh.osv.map_git_origins.is_commit_in_graph(stub: TraversalServiceStub, commit_id: str) → bool[source]#

class swh.osv.map_git_origins.DocProcessor(stub: swh.graph.grpc.swhgraph_pb2_grpc.TraversalServiceStub, session: aiohttp.client.ClientSession, origin_denormalizer: swh.osv.map_origins.OriginDenormalizer, found: set[str] = <factory>, unmapped: set[str] = <factory>, not_found: set[str] = <factory>, found_git_repos: set[str] = <factory>, not_found_git_repos: set[str] = <factory>, not_found_git_repos_by_domain: collections.defaultdict[str, int] = <factory>, found_git_commits: set[str] = <factory>, found_git_commits_not_in_graph: set[str] = <factory>, not_found_git_commits: set[str] = <factory>, not_found_git_commits_by_domain: collections.defaultdict[str, int] = <factory>, not_found_git_commits_by_origin: collections.defaultdict[str, int] = <factory>, invalid_git_commit_ids: set[str] = <factory>)[source]#

Bases: object

stub: TraversalServiceStub#

session: ClientSession#

origin_denormalizer: OriginDenormalizer#

found: set[str]#

unmapped: set[str]#

not_found: set[str]#

total_found = 0#

total_unmapped = 0#

total_not_found = 0#

found_git_repos: set[str]#

not_found_git_repos: set[str]#

not_found_git_repos_by_domain: defaultdict[str, int]#

total_found_git_repos = 0#

total_not_found_git_repos = 0#

found_git_commits: set[str]#

found_git_commits_not_in_graph: set[str]#

not_found_git_commits: set[str]#

not_found_git_commits_by_domain: defaultdict[str, int]#

not_found_git_commits_by_origin: defaultdict[str, int]#

total_found_git_commits = 0#

total_found_git_commits_not_in_graph = 0#

total_not_found_git_commits = 0#

invalid_git_commit_ids: set[str]#

total_invalid_git_commit_ids = 0#

async process_docs(arg)[source]#

async process_git_range(range_: dict) → None[source]#

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