Source code for swh.provenance.grpc_client

# Copyright (C) 2025  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 typing import List, Optional

import grpc

from swh.model.swhids import CoreSWHID, QualifiedSWHID
from swh.provenance.grpc.swhprovenance_pb2 import WhereAreOneRequest, WhereIsOneRequest
from swh.provenance.grpc.swhprovenance_pb2_grpc import ProvenanceServiceStub


[docs] class GrpcProvenance: def __init__(self, url: str): self._channel = grpc.insecure_channel(url) self._stub = ProvenanceServiceStub(self._channel)
[docs] def check_config(self) -> bool: # if the constructor successfully connected, it means we are good return True
[docs] def whereis(self, *, swhid: CoreSWHID) -> Optional[QualifiedSWHID]: str_swhid = str(swhid) result = self._stub.WhereIsOne(WhereIsOneRequest(swhid=str_swhid)) if result is None or result.anchor is None: return None else: assert result.swhid == str_swhid return QualifiedSWHID( object_type=swhid.object_type, object_id=swhid.object_id, anchor=CoreSWHID.from_string(result.anchor), origin=result.origin or None, )
[docs] def whereare(self, *, swhids: List[CoreSWHID]) -> List[Optional[QualifiedSWHID]]: results: List[Optional[QualifiedSWHID]] = [] for result in self._stub.WhereAreOne( WhereAreOneRequest(swhid=list(map(str, swhids))) ): if result is None or result.anchor is None: results.append(None) else: swhid = CoreSWHID.from_string(result.swhid) results.append( QualifiedSWHID( object_type=swhid.object_type, object_id=swhid.object_id, anchor=result.anchor, origin=result.origin or None, ) ) return results