diff --git a/swh/graph/rpc/swhgraph_pb2.pyi b/swh/graph/rpc/swhgraph_pb2.pyi index 140f704..1d099bf 100644 --- a/swh/graph/rpc/swhgraph_pb2.pyi +++ b/swh/graph/rpc/swhgraph_pb2.pyi @@ -1,652 +1,685 @@ """ @generated by mypy-protobuf. Do not edit manually! isort:skip_file """ import builtins import google.protobuf.descriptor import google.protobuf.field_mask_pb2 import google.protobuf.internal.containers import google.protobuf.internal.enum_type_wrapper import google.protobuf.message import typing import typing_extensions DESCRIPTOR: google.protobuf.descriptor.FileDescriptor class _GraphDirection: ValueType = typing.NewType('ValueType', builtins.int) V: typing_extensions.TypeAlias = ValueType class _GraphDirectionEnumTypeWrapper(google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[_GraphDirection.ValueType], builtins.type): DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor FORWARD: _GraphDirection.ValueType # 0 + """Forward DAG: ori -> snp -> rel -> rev -> dir -> cnt""" + BACKWARD: _GraphDirection.ValueType # 1 + """Transposed DAG: cnt -> dir -> rev -> rel -> snp -> ori""" + class GraphDirection(_GraphDirection, metaclass=_GraphDirectionEnumTypeWrapper): + """Direction of the graph""" pass FORWARD: GraphDirection.ValueType # 0 +"""Forward DAG: ori -> snp -> rel -> rev -> dir -> cnt""" + BACKWARD: GraphDirection.ValueType # 1 +"""Transposed DAG: cnt -> dir -> rev -> rel -> snp -> ori""" + global___GraphDirection = GraphDirection class GetNodeRequest(google.protobuf.message.Message): """Describe a node to return""" DESCRIPTOR: google.protobuf.descriptor.Descriptor SWHID_FIELD_NUMBER: builtins.int MASK_FIELD_NUMBER: builtins.int swhid: typing.Text """SWHID of the node to return""" @property def mask(self) -> google.protobuf.field_mask_pb2.FieldMask: """FieldMask of which fields are to be returned (e.g., "swhid,cnt.length"). By default, all fields are returned. """ pass def __init__(self, *, swhid: typing.Text = ..., mask: typing.Optional[google.protobuf.field_mask_pb2.FieldMask] = ..., ) -> None: ... def HasField(self, field_name: typing_extensions.Literal["_mask",b"_mask","mask",b"mask"]) -> builtins.bool: ... def ClearField(self, field_name: typing_extensions.Literal["_mask",b"_mask","mask",b"mask","swhid",b"swhid"]) -> None: ... def WhichOneof(self, oneof_group: typing_extensions.Literal["_mask",b"_mask"]) -> typing.Optional[typing_extensions.Literal["mask"]]: ... global___GetNodeRequest = GetNodeRequest class TraversalRequest(google.protobuf.message.Message): """TraversalRequest describes how a breadth-first traversal should be performed, and what should be returned to the client. """ DESCRIPTOR: google.protobuf.descriptor.Descriptor SRC_FIELD_NUMBER: builtins.int DIRECTION_FIELD_NUMBER: builtins.int EDGES_FIELD_NUMBER: builtins.int MAX_EDGES_FIELD_NUMBER: builtins.int MIN_DEPTH_FIELD_NUMBER: builtins.int MAX_DEPTH_FIELD_NUMBER: builtins.int RETURN_NODES_FIELD_NUMBER: builtins.int MASK_FIELD_NUMBER: builtins.int @property def src(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[typing.Text]: """Set of source nodes (SWHIDs)""" pass direction: global___GraphDirection.ValueType """Direction of the graph to traverse. Defaults to FORWARD.""" edges: typing.Text """Edge restriction string (e.g. "rev:dir,dir:cnt"). Defaults to "*" (all). """ max_edges: builtins.int """Maximum number of edges accessed in the traversal, after which it stops. Defaults to infinite. """ min_depth: builtins.int """Do not return nodes with a depth lower than this number. By default, all depths are returned. """ max_depth: builtins.int """Maximum depth of the traversal, after which it stops. Defaults to infinite. """ @property def return_nodes(self) -> global___NodeFilter: """Filter which nodes will be sent to the stream. By default, all nodes are returned. """ pass @property def mask(self) -> google.protobuf.field_mask_pb2.FieldMask: """FieldMask of which fields are to be returned (e.g., "swhid,cnt.length"). By default, all fields are returned. """ pass def __init__(self, *, src: typing.Optional[typing.Iterable[typing.Text]] = ..., direction: global___GraphDirection.ValueType = ..., edges: typing.Optional[typing.Text] = ..., max_edges: typing.Optional[builtins.int] = ..., min_depth: typing.Optional[builtins.int] = ..., max_depth: typing.Optional[builtins.int] = ..., return_nodes: typing.Optional[global___NodeFilter] = ..., mask: typing.Optional[google.protobuf.field_mask_pb2.FieldMask] = ..., ) -> None: ... def HasField(self, field_name: typing_extensions.Literal["_edges",b"_edges","_mask",b"_mask","_max_depth",b"_max_depth","_max_edges",b"_max_edges","_min_depth",b"_min_depth","_return_nodes",b"_return_nodes","edges",b"edges","mask",b"mask","max_depth",b"max_depth","max_edges",b"max_edges","min_depth",b"min_depth","return_nodes",b"return_nodes"]) -> builtins.bool: ... def ClearField(self, field_name: typing_extensions.Literal["_edges",b"_edges","_mask",b"_mask","_max_depth",b"_max_depth","_max_edges",b"_max_edges","_min_depth",b"_min_depth","_return_nodes",b"_return_nodes","direction",b"direction","edges",b"edges","mask",b"mask","max_depth",b"max_depth","max_edges",b"max_edges","min_depth",b"min_depth","return_nodes",b"return_nodes","src",b"src"]) -> None: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_edges",b"_edges"]) -> typing.Optional[typing_extensions.Literal["edges"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_mask",b"_mask"]) -> typing.Optional[typing_extensions.Literal["mask"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_max_depth",b"_max_depth"]) -> typing.Optional[typing_extensions.Literal["max_depth"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_max_edges",b"_max_edges"]) -> typing.Optional[typing_extensions.Literal["max_edges"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_min_depth",b"_min_depth"]) -> typing.Optional[typing_extensions.Literal["min_depth"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_return_nodes",b"_return_nodes"]) -> typing.Optional[typing_extensions.Literal["return_nodes"]]: ... global___TraversalRequest = TraversalRequest class FindPathToRequest(google.protobuf.message.Message): - """FindPathToRequest describes a request to find the shortest path between a + """FindPathToRequest describes a request to find a shortest path between a set of nodes and a given target criteria, as well as what should be returned in the path. """ DESCRIPTOR: google.protobuf.descriptor.Descriptor SRC_FIELD_NUMBER: builtins.int TARGET_FIELD_NUMBER: builtins.int DIRECTION_FIELD_NUMBER: builtins.int EDGES_FIELD_NUMBER: builtins.int MAX_EDGES_FIELD_NUMBER: builtins.int MAX_DEPTH_FIELD_NUMBER: builtins.int MASK_FIELD_NUMBER: builtins.int @property def src(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[typing.Text]: """Set of source nodes (SWHIDs)""" pass @property def target(self) -> global___NodeFilter: """Target criteria, i.e., what constitutes a valid path destination.""" pass direction: global___GraphDirection.ValueType """Direction of the graph to traverse. Defaults to FORWARD.""" edges: typing.Text """Edge restriction string (e.g. "rev:dir,dir:cnt"). Defaults to "*" (all). """ max_edges: builtins.int """Maximum number of edges accessed in the traversal, after which it stops. Defaults to infinite. """ max_depth: builtins.int """Maximum depth of the traversal, after which it stops. Defaults to infinite. """ @property def mask(self) -> google.protobuf.field_mask_pb2.FieldMask: """FieldMask of which fields are to be returned (e.g., "swhid,cnt.length"). By default, all fields are returned. """ pass def __init__(self, *, src: typing.Optional[typing.Iterable[typing.Text]] = ..., target: typing.Optional[global___NodeFilter] = ..., direction: global___GraphDirection.ValueType = ..., edges: typing.Optional[typing.Text] = ..., max_edges: typing.Optional[builtins.int] = ..., max_depth: typing.Optional[builtins.int] = ..., mask: typing.Optional[google.protobuf.field_mask_pb2.FieldMask] = ..., ) -> None: ... def HasField(self, field_name: typing_extensions.Literal["_edges",b"_edges","_mask",b"_mask","_max_depth",b"_max_depth","_max_edges",b"_max_edges","edges",b"edges","mask",b"mask","max_depth",b"max_depth","max_edges",b"max_edges","target",b"target"]) -> builtins.bool: ... def ClearField(self, field_name: typing_extensions.Literal["_edges",b"_edges","_mask",b"_mask","_max_depth",b"_max_depth","_max_edges",b"_max_edges","direction",b"direction","edges",b"edges","mask",b"mask","max_depth",b"max_depth","max_edges",b"max_edges","src",b"src","target",b"target"]) -> None: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_edges",b"_edges"]) -> typing.Optional[typing_extensions.Literal["edges"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_mask",b"_mask"]) -> typing.Optional[typing_extensions.Literal["mask"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_max_depth",b"_max_depth"]) -> typing.Optional[typing_extensions.Literal["max_depth"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_max_edges",b"_max_edges"]) -> typing.Optional[typing_extensions.Literal["max_edges"]]: ... global___FindPathToRequest = FindPathToRequest class FindPathBetweenRequest(google.protobuf.message.Message): - """FindPathToRequest describes a request to find the shortest path between a + """FindPathToRequest describes a request to find a shortest path between a set of source nodes and a set of destination nodes. It works by performing a bidirectional breadth-first traversal from both sets at the same time. """ DESCRIPTOR: google.protobuf.descriptor.Descriptor SRC_FIELD_NUMBER: builtins.int DST_FIELD_NUMBER: builtins.int DIRECTION_FIELD_NUMBER: builtins.int DIRECTION_REVERSE_FIELD_NUMBER: builtins.int EDGES_FIELD_NUMBER: builtins.int EDGES_REVERSE_FIELD_NUMBER: builtins.int MAX_EDGES_FIELD_NUMBER: builtins.int MAX_DEPTH_FIELD_NUMBER: builtins.int MASK_FIELD_NUMBER: builtins.int @property def src(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[typing.Text]: """Set of source nodes (SWHIDs)""" pass @property def dst(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[typing.Text]: """Set of destination nodes (SWHIDs)""" pass direction: global___GraphDirection.ValueType """Direction of the graph to traverse from the source set. Defaults to FORWARD. """ direction_reverse: global___GraphDirection.ValueType """Direction of the graph to traverse from the destination set. Defaults to the opposite of `direction`. If direction and direction_reverse are identical, it will find the first common successor of both sets in the given direction. """ edges: typing.Text """Edge restriction string for the traversal from the source set. (e.g. "rev:dir,dir:cnt"). Defaults to "*" (all). """ edges_reverse: typing.Text """Edge restriction string for the reverse traversal from the destination set. If not specified: - If `edges` is not specified either, defaults to "*" - If direction == direction_reverse, defaults to `edges` - If direction != direction_reverse, defaults to the reverse of `edges` (e.g. "rev:dir" becomes "dir:rev"). """ max_edges: builtins.int """Maximum number of edges accessed in the traversal, after which it stops. Defaults to infinite. """ max_depth: builtins.int """Maximum depth of the traversal, after which it stops. Defaults to infinite. """ @property def mask(self) -> google.protobuf.field_mask_pb2.FieldMask: """FieldMask of which fields are to be returned (e.g., "swhid,cnt.length"). By default, all fields are returned. """ pass def __init__(self, *, src: typing.Optional[typing.Iterable[typing.Text]] = ..., dst: typing.Optional[typing.Iterable[typing.Text]] = ..., direction: global___GraphDirection.ValueType = ..., direction_reverse: typing.Optional[global___GraphDirection.ValueType] = ..., edges: typing.Optional[typing.Text] = ..., edges_reverse: typing.Optional[typing.Text] = ..., max_edges: typing.Optional[builtins.int] = ..., max_depth: typing.Optional[builtins.int] = ..., mask: typing.Optional[google.protobuf.field_mask_pb2.FieldMask] = ..., ) -> None: ... def HasField(self, field_name: typing_extensions.Literal["_direction_reverse",b"_direction_reverse","_edges",b"_edges","_edges_reverse",b"_edges_reverse","_mask",b"_mask","_max_depth",b"_max_depth","_max_edges",b"_max_edges","direction_reverse",b"direction_reverse","edges",b"edges","edges_reverse",b"edges_reverse","mask",b"mask","max_depth",b"max_depth","max_edges",b"max_edges"]) -> builtins.bool: ... def ClearField(self, field_name: typing_extensions.Literal["_direction_reverse",b"_direction_reverse","_edges",b"_edges","_edges_reverse",b"_edges_reverse","_mask",b"_mask","_max_depth",b"_max_depth","_max_edges",b"_max_edges","direction",b"direction","direction_reverse",b"direction_reverse","dst",b"dst","edges",b"edges","edges_reverse",b"edges_reverse","mask",b"mask","max_depth",b"max_depth","max_edges",b"max_edges","src",b"src"]) -> None: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_direction_reverse",b"_direction_reverse"]) -> typing.Optional[typing_extensions.Literal["direction_reverse"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_edges",b"_edges"]) -> typing.Optional[typing_extensions.Literal["edges"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_edges_reverse",b"_edges_reverse"]) -> typing.Optional[typing_extensions.Literal["edges_reverse"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_mask",b"_mask"]) -> typing.Optional[typing_extensions.Literal["mask"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_max_depth",b"_max_depth"]) -> typing.Optional[typing_extensions.Literal["max_depth"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_max_edges",b"_max_edges"]) -> typing.Optional[typing_extensions.Literal["max_edges"]]: ... global___FindPathBetweenRequest = FindPathBetweenRequest class NodeFilter(google.protobuf.message.Message): - """Represents various criteria that make a given node is "valid". A node is + """Represents various criteria that make a given node "valid". A node is only valid if all the subcriteria present in this message are fulfilled. """ DESCRIPTOR: google.protobuf.descriptor.Descriptor TYPES_FIELD_NUMBER: builtins.int MIN_TRAVERSAL_SUCCESSORS_FIELD_NUMBER: builtins.int MAX_TRAVERSAL_SUCCESSORS_FIELD_NUMBER: builtins.int types: typing.Text """Node restriction string. (e.g. "dir,cnt,rev"). Defaults to "*" (all).""" min_traversal_successors: builtins.int """Minimum number of successors encountered *during the traversal*. Default: no constraint """ max_traversal_successors: builtins.int """Maximum number of successors encountered *during the traversal*. Default: no constraint """ def __init__(self, *, types: typing.Optional[typing.Text] = ..., min_traversal_successors: typing.Optional[builtins.int] = ..., max_traversal_successors: typing.Optional[builtins.int] = ..., ) -> None: ... def HasField(self, field_name: typing_extensions.Literal["_max_traversal_successors",b"_max_traversal_successors","_min_traversal_successors",b"_min_traversal_successors","_types",b"_types","max_traversal_successors",b"max_traversal_successors","min_traversal_successors",b"min_traversal_successors","types",b"types"]) -> builtins.bool: ... def ClearField(self, field_name: typing_extensions.Literal["_max_traversal_successors",b"_max_traversal_successors","_min_traversal_successors",b"_min_traversal_successors","_types",b"_types","max_traversal_successors",b"max_traversal_successors","min_traversal_successors",b"min_traversal_successors","types",b"types"]) -> None: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_max_traversal_successors",b"_max_traversal_successors"]) -> typing.Optional[typing_extensions.Literal["max_traversal_successors"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_min_traversal_successors",b"_min_traversal_successors"]) -> typing.Optional[typing_extensions.Literal["min_traversal_successors"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_types",b"_types"]) -> typing.Optional[typing_extensions.Literal["types"]]: ... global___NodeFilter = NodeFilter class Node(google.protobuf.message.Message): """Represents a node in the graph.""" DESCRIPTOR: google.protobuf.descriptor.Descriptor SWHID_FIELD_NUMBER: builtins.int SUCCESSOR_FIELD_NUMBER: builtins.int NUM_SUCCESSORS_FIELD_NUMBER: builtins.int CNT_FIELD_NUMBER: builtins.int REV_FIELD_NUMBER: builtins.int REL_FIELD_NUMBER: builtins.int ORI_FIELD_NUMBER: builtins.int swhid: typing.Text """The SWHID of the graph node.""" @property def successor(self) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[global___Successor]: """List of relevant successors of this node.""" pass num_successors: builtins.int """Number of relevant successors.""" @property def cnt(self) -> global___ContentData: ... @property def rev(self) -> global___RevisionData: ... @property def rel(self) -> global___ReleaseData: ... @property def ori(self) -> global___OriginData: ... def __init__(self, *, swhid: typing.Text = ..., successor: typing.Optional[typing.Iterable[global___Successor]] = ..., num_successors: typing.Optional[builtins.int] = ..., cnt: typing.Optional[global___ContentData] = ..., rev: typing.Optional[global___RevisionData] = ..., rel: typing.Optional[global___ReleaseData] = ..., ori: typing.Optional[global___OriginData] = ..., ) -> None: ... def HasField(self, field_name: typing_extensions.Literal["_num_successors",b"_num_successors","cnt",b"cnt","data",b"data","num_successors",b"num_successors","ori",b"ori","rel",b"rel","rev",b"rev"]) -> builtins.bool: ... def ClearField(self, field_name: typing_extensions.Literal["_num_successors",b"_num_successors","cnt",b"cnt","data",b"data","num_successors",b"num_successors","ori",b"ori","rel",b"rel","rev",b"rev","successor",b"successor","swhid",b"swhid"]) -> None: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_num_successors",b"_num_successors"]) -> typing.Optional[typing_extensions.Literal["num_successors"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["data",b"data"]) -> typing.Optional[typing_extensions.Literal["cnt","rev","rel","ori"]]: ... global___Node = Node class Path(google.protobuf.message.Message): """Represents a path in the graph.""" DESCRIPTOR: google.protobuf.descriptor.Descriptor NODE_FIELD_NUMBER: builtins.int MIDPOINT_INDEX_FIELD_NUMBER: builtins.int @property def node(self) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[global___Node]: """List of nodes in the path, from source to destination""" pass midpoint_index: builtins.int """Index of the "midpoint" of the path. For paths obtained with bidirectional search queries, this is the node that joined the two sets together. When looking for a common ancestor between two nodes by performing a FindPathBetween search with two backward graphs, this will be the index of the common ancestor in the path. """ def __init__(self, *, node: typing.Optional[typing.Iterable[global___Node]] = ..., midpoint_index: typing.Optional[builtins.int] = ..., ) -> None: ... def HasField(self, field_name: typing_extensions.Literal["_midpoint_index",b"_midpoint_index","midpoint_index",b"midpoint_index"]) -> builtins.bool: ... def ClearField(self, field_name: typing_extensions.Literal["_midpoint_index",b"_midpoint_index","midpoint_index",b"midpoint_index","node",b"node"]) -> None: ... def WhichOneof(self, oneof_group: typing_extensions.Literal["_midpoint_index",b"_midpoint_index"]) -> typing.Optional[typing_extensions.Literal["midpoint_index"]]: ... global___Path = Path class Successor(google.protobuf.message.Message): """Represents a successor of a given node.""" DESCRIPTOR: google.protobuf.descriptor.Descriptor SWHID_FIELD_NUMBER: builtins.int LABEL_FIELD_NUMBER: builtins.int swhid: typing.Text """The SWHID of the successor""" @property def label(self) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[global___EdgeLabel]: """A list of edge labels for the given edge""" pass def __init__(self, *, swhid: typing.Optional[typing.Text] = ..., label: typing.Optional[typing.Iterable[global___EdgeLabel]] = ..., ) -> None: ... def HasField(self, field_name: typing_extensions.Literal["_swhid",b"_swhid","swhid",b"swhid"]) -> builtins.bool: ... def ClearField(self, field_name: typing_extensions.Literal["_swhid",b"_swhid","label",b"label","swhid",b"swhid"]) -> None: ... def WhichOneof(self, oneof_group: typing_extensions.Literal["_swhid",b"_swhid"]) -> typing.Optional[typing_extensions.Literal["swhid"]]: ... global___Successor = Successor class ContentData(google.protobuf.message.Message): """Content node properties""" DESCRIPTOR: google.protobuf.descriptor.Descriptor LENGTH_FIELD_NUMBER: builtins.int IS_SKIPPED_FIELD_NUMBER: builtins.int length: builtins.int """Length of the blob, in bytes""" is_skipped: builtins.bool """Whether the content was skipped during ingestion.""" def __init__(self, *, length: typing.Optional[builtins.int] = ..., is_skipped: typing.Optional[builtins.bool] = ..., ) -> None: ... def HasField(self, field_name: typing_extensions.Literal["_is_skipped",b"_is_skipped","_length",b"_length","is_skipped",b"is_skipped","length",b"length"]) -> builtins.bool: ... def ClearField(self, field_name: typing_extensions.Literal["_is_skipped",b"_is_skipped","_length",b"_length","is_skipped",b"is_skipped","length",b"length"]) -> None: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_is_skipped",b"_is_skipped"]) -> typing.Optional[typing_extensions.Literal["is_skipped"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_length",b"_length"]) -> typing.Optional[typing_extensions.Literal["length"]]: ... global___ContentData = ContentData class RevisionData(google.protobuf.message.Message): """Revision node properties""" DESCRIPTOR: google.protobuf.descriptor.Descriptor AUTHOR_FIELD_NUMBER: builtins.int AUTHOR_DATE_FIELD_NUMBER: builtins.int AUTHOR_DATE_OFFSET_FIELD_NUMBER: builtins.int COMMITTER_FIELD_NUMBER: builtins.int COMMITTER_DATE_FIELD_NUMBER: builtins.int COMMITTER_DATE_OFFSET_FIELD_NUMBER: builtins.int MESSAGE_FIELD_NUMBER: builtins.int author: builtins.int """Revision author ID (anonymized)""" author_date: builtins.int """UNIX timestamp of the revision date (UTC)""" author_date_offset: builtins.int """Timezone of the revision author date as an offset from UTC""" committer: builtins.int """Revision committer ID (anonymized)""" committer_date: builtins.int """UNIX timestamp of the revision committer date (UTC)""" committer_date_offset: builtins.int """Timezone of the revision committer date as an offset from UTC""" message: builtins.bytes """Revision message""" def __init__(self, *, author: typing.Optional[builtins.int] = ..., author_date: typing.Optional[builtins.int] = ..., author_date_offset: typing.Optional[builtins.int] = ..., committer: typing.Optional[builtins.int] = ..., committer_date: typing.Optional[builtins.int] = ..., committer_date_offset: typing.Optional[builtins.int] = ..., message: typing.Optional[builtins.bytes] = ..., ) -> None: ... def HasField(self, field_name: typing_extensions.Literal["_author",b"_author","_author_date",b"_author_date","_author_date_offset",b"_author_date_offset","_committer",b"_committer","_committer_date",b"_committer_date","_committer_date_offset",b"_committer_date_offset","_message",b"_message","author",b"author","author_date",b"author_date","author_date_offset",b"author_date_offset","committer",b"committer","committer_date",b"committer_date","committer_date_offset",b"committer_date_offset","message",b"message"]) -> builtins.bool: ... def ClearField(self, field_name: typing_extensions.Literal["_author",b"_author","_author_date",b"_author_date","_author_date_offset",b"_author_date_offset","_committer",b"_committer","_committer_date",b"_committer_date","_committer_date_offset",b"_committer_date_offset","_message",b"_message","author",b"author","author_date",b"author_date","author_date_offset",b"author_date_offset","committer",b"committer","committer_date",b"committer_date","committer_date_offset",b"committer_date_offset","message",b"message"]) -> None: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_author",b"_author"]) -> typing.Optional[typing_extensions.Literal["author"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_author_date",b"_author_date"]) -> typing.Optional[typing_extensions.Literal["author_date"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_author_date_offset",b"_author_date_offset"]) -> typing.Optional[typing_extensions.Literal["author_date_offset"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_committer",b"_committer"]) -> typing.Optional[typing_extensions.Literal["committer"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_committer_date",b"_committer_date"]) -> typing.Optional[typing_extensions.Literal["committer_date"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_committer_date_offset",b"_committer_date_offset"]) -> typing.Optional[typing_extensions.Literal["committer_date_offset"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_message",b"_message"]) -> typing.Optional[typing_extensions.Literal["message"]]: ... global___RevisionData = RevisionData class ReleaseData(google.protobuf.message.Message): """Release node properties""" DESCRIPTOR: google.protobuf.descriptor.Descriptor AUTHOR_FIELD_NUMBER: builtins.int AUTHOR_DATE_FIELD_NUMBER: builtins.int AUTHOR_DATE_OFFSET_FIELD_NUMBER: builtins.int NAME_FIELD_NUMBER: builtins.int MESSAGE_FIELD_NUMBER: builtins.int author: builtins.int """Release author ID (anonymized)""" author_date: builtins.int """UNIX timestamp of the release date (UTC)""" author_date_offset: builtins.int """Timezone of the release author date as an offset from UTC""" name: builtins.bytes """Release name""" message: builtins.bytes """Release message""" def __init__(self, *, author: typing.Optional[builtins.int] = ..., author_date: typing.Optional[builtins.int] = ..., author_date_offset: typing.Optional[builtins.int] = ..., name: typing.Optional[builtins.bytes] = ..., message: typing.Optional[builtins.bytes] = ..., ) -> None: ... def HasField(self, field_name: typing_extensions.Literal["_author",b"_author","_author_date",b"_author_date","_author_date_offset",b"_author_date_offset","_message",b"_message","_name",b"_name","author",b"author","author_date",b"author_date","author_date_offset",b"author_date_offset","message",b"message","name",b"name"]) -> builtins.bool: ... def ClearField(self, field_name: typing_extensions.Literal["_author",b"_author","_author_date",b"_author_date","_author_date_offset",b"_author_date_offset","_message",b"_message","_name",b"_name","author",b"author","author_date",b"author_date","author_date_offset",b"author_date_offset","message",b"message","name",b"name"]) -> None: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_author",b"_author"]) -> typing.Optional[typing_extensions.Literal["author"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_author_date",b"_author_date"]) -> typing.Optional[typing_extensions.Literal["author_date"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_author_date_offset",b"_author_date_offset"]) -> typing.Optional[typing_extensions.Literal["author_date_offset"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_message",b"_message"]) -> typing.Optional[typing_extensions.Literal["message"]]: ... @typing.overload def WhichOneof(self, oneof_group: typing_extensions.Literal["_name",b"_name"]) -> typing.Optional[typing_extensions.Literal["name"]]: ... global___ReleaseData = ReleaseData class OriginData(google.protobuf.message.Message): """Origin node properties""" DESCRIPTOR: google.protobuf.descriptor.Descriptor URL_FIELD_NUMBER: builtins.int url: typing.Text """URL of the origin""" def __init__(self, *, url: typing.Optional[typing.Text] = ..., ) -> None: ... def HasField(self, field_name: typing_extensions.Literal["_url",b"_url","url",b"url"]) -> builtins.bool: ... def ClearField(self, field_name: typing_extensions.Literal["_url",b"_url","url",b"url"]) -> None: ... def WhichOneof(self, oneof_group: typing_extensions.Literal["_url",b"_url"]) -> typing.Optional[typing_extensions.Literal["url"]]: ... global___OriginData = OriginData class EdgeLabel(google.protobuf.message.Message): DESCRIPTOR: google.protobuf.descriptor.Descriptor NAME_FIELD_NUMBER: builtins.int PERMISSION_FIELD_NUMBER: builtins.int name: builtins.bytes """Directory entry name for directories, branch name for snapshots""" permission: builtins.int """Entry permission (only set for directories).""" def __init__(self, *, name: builtins.bytes = ..., permission: builtins.int = ..., ) -> None: ... def ClearField(self, field_name: typing_extensions.Literal["name",b"name","permission",b"permission"]) -> None: ... global___EdgeLabel = EdgeLabel class CountResponse(google.protobuf.message.Message): DESCRIPTOR: google.protobuf.descriptor.Descriptor COUNT_FIELD_NUMBER: builtins.int count: builtins.int def __init__(self, *, count: builtins.int = ..., ) -> None: ... def ClearField(self, field_name: typing_extensions.Literal["count",b"count"]) -> None: ... global___CountResponse = CountResponse class StatsRequest(google.protobuf.message.Message): DESCRIPTOR: google.protobuf.descriptor.Descriptor def __init__(self, ) -> None: ... global___StatsRequest = StatsRequest class StatsResponse(google.protobuf.message.Message): DESCRIPTOR: google.protobuf.descriptor.Descriptor NUM_NODES_FIELD_NUMBER: builtins.int NUM_EDGES_FIELD_NUMBER: builtins.int COMPRESSION_RATIO_FIELD_NUMBER: builtins.int BITS_PER_NODE_FIELD_NUMBER: builtins.int BITS_PER_EDGE_FIELD_NUMBER: builtins.int AVG_LOCALITY_FIELD_NUMBER: builtins.int INDEGREE_MIN_FIELD_NUMBER: builtins.int INDEGREE_MAX_FIELD_NUMBER: builtins.int INDEGREE_AVG_FIELD_NUMBER: builtins.int OUTDEGREE_MIN_FIELD_NUMBER: builtins.int OUTDEGREE_MAX_FIELD_NUMBER: builtins.int OUTDEGREE_AVG_FIELD_NUMBER: builtins.int num_nodes: builtins.int """Number of nodes in the graph""" num_edges: builtins.int """Number of edges in the graph""" compression_ratio: builtins.float + """Ratio between the graph size and the information-theoretical lower + bound + """ + bits_per_node: builtins.float + """Number of bits per node (overall graph size in bits divided by the + number of nodes) + """ + bits_per_edge: builtins.float + """Number of bits per edge (overall graph size in bits divided by the + number of arcs). + """ + avg_locality: builtins.float indegree_min: builtins.int + """Smallest indegree""" + indegree_max: builtins.int + """Largest indegree""" + indegree_avg: builtins.float + """Average indegree""" + outdegree_min: builtins.int + """Smallest outdegree""" + outdegree_max: builtins.int + """Largest outdegree""" + outdegree_avg: builtins.float + """Average outdegree""" + def __init__(self, *, num_nodes: builtins.int = ..., num_edges: builtins.int = ..., compression_ratio: builtins.float = ..., bits_per_node: builtins.float = ..., bits_per_edge: builtins.float = ..., avg_locality: builtins.float = ..., indegree_min: builtins.int = ..., indegree_max: builtins.int = ..., indegree_avg: builtins.float = ..., outdegree_min: builtins.int = ..., outdegree_max: builtins.int = ..., outdegree_avg: builtins.float = ..., ) -> None: ... def ClearField(self, field_name: typing_extensions.Literal["avg_locality",b"avg_locality","bits_per_edge",b"bits_per_edge","bits_per_node",b"bits_per_node","compression_ratio",b"compression_ratio","indegree_avg",b"indegree_avg","indegree_max",b"indegree_max","indegree_min",b"indegree_min","num_edges",b"num_edges","num_nodes",b"num_nodes","outdegree_avg",b"outdegree_avg","outdegree_max",b"outdegree_max","outdegree_min",b"outdegree_min"]) -> None: ... global___StatsResponse = StatsResponse diff --git a/swh/graph/rpc/swhgraph_pb2_grpc.py b/swh/graph/rpc/swhgraph_pb2_grpc.py index 1ca7ce0..ff60903 100644 --- a/swh/graph/rpc/swhgraph_pb2_grpc.py +++ b/swh/graph/rpc/swhgraph_pb2_grpc.py @@ -1,303 +1,303 @@ # Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT! """Client and server classes corresponding to protobuf-defined services.""" import grpc from swh.graph.rpc import swhgraph_pb2 as swh_dot_graph_dot_rpc_dot_swhgraph__pb2 class TraversalServiceStub(object): """Graph traversal service """ def __init__(self, channel): """Constructor. Args: channel: A grpc.Channel. """ self.GetNode = channel.unary_unary( '/swh.graph.TraversalService/GetNode', request_serializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.GetNodeRequest.SerializeToString, response_deserializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.Node.FromString, ) self.Traverse = channel.unary_stream( '/swh.graph.TraversalService/Traverse', request_serializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.TraversalRequest.SerializeToString, response_deserializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.Node.FromString, ) self.FindPathTo = channel.unary_unary( '/swh.graph.TraversalService/FindPathTo', request_serializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.FindPathToRequest.SerializeToString, response_deserializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.Path.FromString, ) self.FindPathBetween = channel.unary_unary( '/swh.graph.TraversalService/FindPathBetween', request_serializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.FindPathBetweenRequest.SerializeToString, response_deserializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.Path.FromString, ) self.CountNodes = channel.unary_unary( '/swh.graph.TraversalService/CountNodes', request_serializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.TraversalRequest.SerializeToString, response_deserializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.CountResponse.FromString, ) self.CountEdges = channel.unary_unary( '/swh.graph.TraversalService/CountEdges', request_serializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.TraversalRequest.SerializeToString, response_deserializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.CountResponse.FromString, ) self.Stats = channel.unary_unary( '/swh.graph.TraversalService/Stats', request_serializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.StatsRequest.SerializeToString, response_deserializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.StatsResponse.FromString, ) class TraversalServiceServicer(object): """Graph traversal service """ def GetNode(self, request, context): """GetNode returns a single Node and its properties. """ context.set_code(grpc.StatusCode.UNIMPLEMENTED) context.set_details('Method not implemented!') raise NotImplementedError('Method not implemented!') def Traverse(self, request, context): """Traverse performs a breadth-first graph traversal from a set of source nodes, then streams the nodes it encounters (if they match a given return filter), along with their properties. """ context.set_code(grpc.StatusCode.UNIMPLEMENTED) context.set_details('Method not implemented!') raise NotImplementedError('Method not implemented!') def FindPathTo(self, request, context): - """FindPathTo searches for the shortest path between a set of source nodes + """FindPathTo searches for a shortest path between a set of source nodes and a node that matches a specific *criteria*. It does so by performing a breadth-first search from the source node, until any node that matches the given criteria is found, then follows - back its parents to return the shortest path from the source set to that + back its parents to return a shortest path from the source set to that node. """ context.set_code(grpc.StatusCode.UNIMPLEMENTED) context.set_details('Method not implemented!') raise NotImplementedError('Method not implemented!') def FindPathBetween(self, request, context): - """FindPathBetween searches for the shortest path between a set of source + """FindPathBetween searches for a shortest path between a set of source nodes and a set of destination nodes. It does so by performing a *bidirectional breadth-first search*, i.e., two parallel breadth-first searches, one from the source set ("src-BFS") and one from the destination set ("dst-BFS"), until both searches find a common node that joins their visited sets. This node is called the "midpoint node". - The path returned is the path src -> ... -> midpoint * -> ... -> dst, - which is the shortest path between src and dst. + The path returned is the path src -> ... -> midpoint -> ... -> dst, + which is always a shortest path between src and dst. The graph direction of both BFS can be configured separately. By default, the dst-BFS will use the graph in the opposite direction than the src-BFS (if direction = FORWARD, by default direction_reverse = BACKWARD, and vice-versa). The default behavior is thus to search for - the shortest path between two nodes in a given direction. However, one + a shortest path between two nodes in a given direction. However, one can also specify FORWARD or BACKWARD for *both* the src-BFS and the dst-BFS. This will search for a common descendant or a common ancestor between the two sets, respectively. These will be the midpoints of the returned path. """ context.set_code(grpc.StatusCode.UNIMPLEMENTED) context.set_details('Method not implemented!') raise NotImplementedError('Method not implemented!') def CountNodes(self, request, context): """CountNodes does the same as Traverse, but only returns the number of nodes accessed during the traversal. """ context.set_code(grpc.StatusCode.UNIMPLEMENTED) context.set_details('Method not implemented!') raise NotImplementedError('Method not implemented!') def CountEdges(self, request, context): """CountEdges does the same as Traverse, but only returns the number of edges accessed during the traversal. """ context.set_code(grpc.StatusCode.UNIMPLEMENTED) context.set_details('Method not implemented!') raise NotImplementedError('Method not implemented!') def Stats(self, request, context): - """Stats returns various + """Stats returns various statistics on the overall graph. """ context.set_code(grpc.StatusCode.UNIMPLEMENTED) context.set_details('Method not implemented!') raise NotImplementedError('Method not implemented!') def add_TraversalServiceServicer_to_server(servicer, server): rpc_method_handlers = { 'GetNode': grpc.unary_unary_rpc_method_handler( servicer.GetNode, request_deserializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.GetNodeRequest.FromString, response_serializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.Node.SerializeToString, ), 'Traverse': grpc.unary_stream_rpc_method_handler( servicer.Traverse, request_deserializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.TraversalRequest.FromString, response_serializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.Node.SerializeToString, ), 'FindPathTo': grpc.unary_unary_rpc_method_handler( servicer.FindPathTo, request_deserializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.FindPathToRequest.FromString, response_serializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.Path.SerializeToString, ), 'FindPathBetween': grpc.unary_unary_rpc_method_handler( servicer.FindPathBetween, request_deserializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.FindPathBetweenRequest.FromString, response_serializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.Path.SerializeToString, ), 'CountNodes': grpc.unary_unary_rpc_method_handler( servicer.CountNodes, request_deserializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.TraversalRequest.FromString, response_serializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.CountResponse.SerializeToString, ), 'CountEdges': grpc.unary_unary_rpc_method_handler( servicer.CountEdges, request_deserializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.TraversalRequest.FromString, response_serializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.CountResponse.SerializeToString, ), 'Stats': grpc.unary_unary_rpc_method_handler( servicer.Stats, request_deserializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.StatsRequest.FromString, response_serializer=swh_dot_graph_dot_rpc_dot_swhgraph__pb2.StatsResponse.SerializeToString, ), } generic_handler = grpc.method_handlers_generic_handler( 'swh.graph.TraversalService', rpc_method_handlers) server.add_generic_rpc_handlers((generic_handler,)) # This class is part of an EXPERIMENTAL API. class TraversalService(object): """Graph traversal service """ @staticmethod def GetNode(request, target, options=(), channel_credentials=None, call_credentials=None, insecure=False, compression=None, wait_for_ready=None, timeout=None, metadata=None): return grpc.experimental.unary_unary(request, target, '/swh.graph.TraversalService/GetNode', swh_dot_graph_dot_rpc_dot_swhgraph__pb2.GetNodeRequest.SerializeToString, swh_dot_graph_dot_rpc_dot_swhgraph__pb2.Node.FromString, options, channel_credentials, insecure, call_credentials, compression, wait_for_ready, timeout, metadata) @staticmethod def Traverse(request, target, options=(), channel_credentials=None, call_credentials=None, insecure=False, compression=None, wait_for_ready=None, timeout=None, metadata=None): return grpc.experimental.unary_stream(request, target, '/swh.graph.TraversalService/Traverse', swh_dot_graph_dot_rpc_dot_swhgraph__pb2.TraversalRequest.SerializeToString, swh_dot_graph_dot_rpc_dot_swhgraph__pb2.Node.FromString, options, channel_credentials, insecure, call_credentials, compression, wait_for_ready, timeout, metadata) @staticmethod def FindPathTo(request, target, options=(), channel_credentials=None, call_credentials=None, insecure=False, compression=None, wait_for_ready=None, timeout=None, metadata=None): return grpc.experimental.unary_unary(request, target, '/swh.graph.TraversalService/FindPathTo', swh_dot_graph_dot_rpc_dot_swhgraph__pb2.FindPathToRequest.SerializeToString, swh_dot_graph_dot_rpc_dot_swhgraph__pb2.Path.FromString, options, channel_credentials, insecure, call_credentials, compression, wait_for_ready, timeout, metadata) @staticmethod def FindPathBetween(request, target, options=(), channel_credentials=None, call_credentials=None, insecure=False, compression=None, wait_for_ready=None, timeout=None, metadata=None): return grpc.experimental.unary_unary(request, target, '/swh.graph.TraversalService/FindPathBetween', swh_dot_graph_dot_rpc_dot_swhgraph__pb2.FindPathBetweenRequest.SerializeToString, swh_dot_graph_dot_rpc_dot_swhgraph__pb2.Path.FromString, options, channel_credentials, insecure, call_credentials, compression, wait_for_ready, timeout, metadata) @staticmethod def CountNodes(request, target, options=(), channel_credentials=None, call_credentials=None, insecure=False, compression=None, wait_for_ready=None, timeout=None, metadata=None): return grpc.experimental.unary_unary(request, target, '/swh.graph.TraversalService/CountNodes', swh_dot_graph_dot_rpc_dot_swhgraph__pb2.TraversalRequest.SerializeToString, swh_dot_graph_dot_rpc_dot_swhgraph__pb2.CountResponse.FromString, options, channel_credentials, insecure, call_credentials, compression, wait_for_ready, timeout, metadata) @staticmethod def CountEdges(request, target, options=(), channel_credentials=None, call_credentials=None, insecure=False, compression=None, wait_for_ready=None, timeout=None, metadata=None): return grpc.experimental.unary_unary(request, target, '/swh.graph.TraversalService/CountEdges', swh_dot_graph_dot_rpc_dot_swhgraph__pb2.TraversalRequest.SerializeToString, swh_dot_graph_dot_rpc_dot_swhgraph__pb2.CountResponse.FromString, options, channel_credentials, insecure, call_credentials, compression, wait_for_ready, timeout, metadata) @staticmethod def Stats(request, target, options=(), channel_credentials=None, call_credentials=None, insecure=False, compression=None, wait_for_ready=None, timeout=None, metadata=None): return grpc.experimental.unary_unary(request, target, '/swh.graph.TraversalService/Stats', swh_dot_graph_dot_rpc_dot_swhgraph__pb2.StatsRequest.SerializeToString, swh_dot_graph_dot_rpc_dot_swhgraph__pb2.StatsResponse.FromString, options, channel_credentials, insecure, call_credentials, compression, wait_for_ready, timeout, metadata)