class_docstr: Metadata resolver for reasoning about metadata information.
  
  Metadata resolver composes and sends queries to get a lineage graph from
  metadata store. The lineage graph is a snapshot view of the ML pipeline's
  metadata, containing all information needed to answer quetions about the
  lineage of nodes of interest.
  Based on the lineage graph, metadata resolver provides a set of util functions
  that help users reason about metadata information by post-processing the
  graph.
  It can be considered as a wrapper layer built on top of metadata store's graph
  tracing APIs.
  
  Example:
  
  # `store` is a metadata store that has been initialized.
  resolver = MetadataResolver(store)
  # Call functions defined in MetadataResolver. For example:
  artifact_ids = [model.id]
  downstream_artifacts_dict = get_downstream_artifacts_by_artifact_ids(
      artifact_ids, max_num_hops = 2
  )
docstr: Get matched artifacts of a uri and their provenance successor artifacts.
  
  For each artifact matched by the given `artifact_uri`, treat it as a
  starting artifact and get artifacts that are connected to them via a path in
  the downstream direction like:
  artifact_i -> INPUT_event -> execution_j -> OUTPUT_event -> artifact_k.
  
  Args:
      artifact_uri: the uri of starting artifacts. At most 100 artifacts
        matched by the uri are considered as starting artifacts.
      max_num_hops: maximum number of hops performed for downstream tracing. A
        hop is defined as a jump to the next node following the path of node
        -> event -> next_node. For example, in the lineage graph artifact_1 ->
        event -> execution_1 -> event -> artifact_2: artifact_2 is 2 hops away
        from artifact_1, and execution_1 is 1 hop away from artifact_1.
        `max_num_hops` cannot exceed 100 nor be negative.
  
  Returns:
      Mapping of artifact ids to a list of downstream artifacts.
docstr_tok: ["Get","matched","artifacts","of","a","uri","and","their","provenance","successor","artifacts",".","For","each","artifact","matched","by","the","given","`","artifact_uri","`",",","treat","it","as","a","starting","artifact","and","get","artifacts","that","are","connected","to","them","via","a","path","in","the","downstream","direction","like",":","artifact_i","-",">","INPUT_event","-",">","execution_j","-",">","OUTPUT_event","-",">","artifact_k",".","Args",":","artifact_uri",":","the","uri","of","starting","artifacts",".","At","most","100","artifacts","matched","by","the","uri","are","considered","as","starting","artifacts",".","max_num_hops",":","maximum","number","of","hops","performed","for","downstream","tracing",".","A","hop","is","defined","as","a","jump","to","the","next","node","following","the","path","of","node","-",">","event","-",">","next_node",".","For","example",",","in","the","lineage","graph","artifact_1","-",">","event","-",">","execution_1","-",">","event","-",">","artifact_2",":","artifact_2","is","2","hops","away","from","artifact_1",",","and","execution_1","is","1","hop","away","from","artifact_1",".","`","max_num_hops","`","can","not","exceed","100","nor","be","negative",".","Returns",":","Mapping","of","artifact","ids","to","a","list","of","downstream","artifacts","."]
code: def get_downstream_artifacts_by_artifact_uri(
      self, artifact_uri: str, max_num_hops: int = _MAX_NUM_HOPS
  ) -> Dict[int, List[metadata_store_pb2.Artifact]]:
      """Get matched artifacts of a uri and their provenance successor artifacts.
  
      For each artifact matched by the given `artifact_uri`, treat it as a
      starting artifact and get artifacts that are connected to them via a path in
      the downstream direction like:
      artifact_i -> INPUT_event -> execution_j -> OUTPUT_event -> artifact_k.
  
      Args:
          artifact_uri: the uri of starting artifacts. At most 100 artifacts
            matched by the uri are considered as starting artifacts.
          max_num_hops: maximum number of hops performed for downstream tracing. A
            hop is defined as a jump to the next node following the path of node
            -> event -> next_node. For example, in the lineage graph artifact_1 ->
            event -> execution_1 -> event -> artifact_2: artifact_2 is 2 hops away
            from artifact_1, and execution_1 is 1 hop away from artifact_1.
            `max_num_hops` cannot exceed 100 nor be negative.
  
      Returns:
          Mapping of artifact ids to a list of downstream artifacts.
      """
      if not artifact_uri:
          raise ValueError("`artifact_uri` is empty.")
      if max_num_hops > _MAX_NUM_HOPS or max_num_hops < 0:
          raise ValueError(
              "Number of hops is larger than supported or is negative."
          )
  
      starting_artifacts_filter_query = f'uri = "{artifact_uri}"'
  
      options = metadata_store_pb2.LineageSubgraphQueryOptions(
          starting_artifacts=metadata_store_pb2.LineageSubgraphQueryOptions.StartingNodes(
              filter_query=starting_artifacts_filter_query
          ),
          max_num_hops=max_num_hops,
          direction=metadata_store_pb2.LineageSubgraphQueryOptions.Direction.DOWNSTREAM,
      )
      lineage_graph = self._store.get_lineage_subgraph(
          query_options=options,
          field_mask_paths=[
              _ARTIFACTS_FIELD_MASK_PATH,
              _EVENTS_FIELD_MASK_PATH,
          ],
      )
  
      artifact_ids = [
          artifact.id
          for artifact in lineage_graph.artifacts
          if artifact.uri == artifact_uri
      ]
      artifacts_to_subgraph = metadata_resolver_utils.get_subgraphs_by_artifact_ids(
          artifact_ids,
          metadata_store_pb2.LineageSubgraphQueryOptions.Direction.DOWNSTREAM,
          lineage_graph,
      )
      return {
          artifact_id: list(subgraph.artifacts)
          for artifact_id, subgraph in artifacts_to_subgraph.items()
      }
  
file_path: ml-metadata/ml_metadata/tools/mlmd_resolver/metadata_resolver.py
imports: from typing import Callable, Dict, List, Optional
  from ml_metadata import metadata_store
  from ml_metadata.proto import metadata_store_pb2
  from ml_metadata.tools.mlmd_resolver import metadata_resolver_utils
label_desc: Use image node_id 3 for calling the MetadataResolver obj's underlying member method code with example usage: obj.get_downstream_artifacts_by_artifact_uri(artifact_uri, max_num_hops) and returns: unknown