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: Given a list of artifact ids, get their provenance ancestor artifacts.
  
  For each artifact matched by a given `artifact_id`, treat it as a starting
  artifact and get artifacts that are connected to them within `max_num_hops`
  via a path in the upstream direction like:
  artifact_i -> OUTPUT_event -> execution_j -> INPUT_event -> artifact_k.
  
  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.
  
  Args:
      artifact_ids: ids of starting artifacts. At most 100 ids are supported.
        Returns empty result if `artifact_ids` is empty.
      max_num_hops: maximum number of hops performed for upstream tracing.
        `max_num_hops` cannot exceed 100 nor be negative.
      filter_query: a query string filtering upstream artifacts by their own
        attributes or the attributes of immediate neighbors. Please refer to
        go/mlmd-filter-query-guide for more detailed guidance. Note: if
        `filter_query` is specified and `max_num_hops` is 0, it's equivalent
        to getting filtered artifacts by artifact ids with `get_artifacts()`.
      event_filter: an optional callable object for filtering events in the
        paths towards the upstream artifacts. Only an event with
        `event_filter(event)` evaluated to True will be considered as valid
        and kept in the path.
  
  Returns:
  Mapping of artifact ids to a list of upstream artifacts.
docstr_tok: ["Given","a","list","of","artifact","ids",",","get","their","provenance","ancestor","artifacts",".","For","each","artifact","matched","by","a","given","`","artifact_id","`",",","treat","it","as","a","starting","artifact","and","get","artifacts","that","are","connected","to","them","within","`","max_num_hops","`","via","a","path","in","the","upstream","direction","like",":","artifact_i","-",">","OUTPUT_event","-",">","execution_j","-",">","INPUT_event","-",">","artifact_k",".","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",".","Args",":","artifact_ids",":","ids","of","starting","artifacts",".","At","most","100","ids","are","supported",".","Returns","empty","result","if","`","artifact_ids","`","is","empty",".","max_num_hops",":","maximum","number","of","hops","performed","for","upstream","tracing",".","`","max_num_hops","`","can","not","exceed","100","nor","be","negative",".","filter_query",":","a","query","string","filtering","upstream","artifacts","by","their","own","attributes","or","the","attributes","of","immediate","neighbors",".","Please","refer","to","go\/mlmd-filter-query-guide","for","more","detailed","guidance",".","Note",":","if","`","filter_query","`","is","specified","and","`","max_num_hops","`","is","0",",","it","'s","equivalent","to","getting","filtered","artifacts","by","artifact","ids","with","`","get_artifacts","(",")","`",".","event_filter",":","an","optional","callable","object","for","filtering","events","in","the","paths","towards","the","upstream","artifacts",".","Only","an","event","with","`","event_filter","(","event",")","`","evaluated","to","True","will","be","considered","as","valid","and","kept","in","the","path",".","Returns",":","Mapping","of","artifact","ids","to","a","list","of","upstream","artifacts","."]
code: def get_upstream_artifacts_by_artifact_ids(
      self,
      artifact_ids: List[int],
      max_num_hops: int = _MAX_NUM_HOPS,
      filter_query: str = "",
      event_filter: Optional[
          Callable[[metadata_store_pb2.Event], bool]
      ] = None,
  ) -> Dict[int, List[metadata_store_pb2.Artifact]]:
      """Given a list of artifact ids, get their provenance ancestor artifacts.
  
      For each artifact matched by a given `artifact_id`, treat it as a starting
      artifact and get artifacts that are connected to them within `max_num_hops`
      via a path in the upstream direction like:
      artifact_i -> OUTPUT_event -> execution_j -> INPUT_event -> artifact_k.
  
      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.
  
      Args:
          artifact_ids: ids of starting artifacts. At most 100 ids are supported.
            Returns empty result if `artifact_ids` is empty.
          max_num_hops: maximum number of hops performed for upstream tracing.
            `max_num_hops` cannot exceed 100 nor be negative.
          filter_query: a query string filtering upstream artifacts by their own
            attributes or the attributes of immediate neighbors. Please refer to
            go/mlmd-filter-query-guide for more detailed guidance. Note: if
            `filter_query` is specified and `max_num_hops` is 0, it's equivalent
            to getting filtered artifacts by artifact ids with `get_artifacts()`.
          event_filter: an optional callable object for filtering events in the
            paths towards the upstream artifacts. Only an event with
            `event_filter(event)` evaluated to True will be considered as valid
            and kept in the path.
  
      Returns:
      Mapping of artifact ids to a list of upstream artifacts.
      """
      if len(artifact_ids) > _MAX_NUM_STARTING_NODES:
          raise ValueError(
              "Number of artifact ids is larger than supported."
          )
      if not artifact_ids:
          return {}
      if max_num_hops > _MAX_NUM_HOPS or max_num_hops < 0:
          raise ValueError(
              "Number of hops is larger than supported or is negative."
          )
  
      artifact_ids_str = ",".join(str(id) for id in artifact_ids)
      # If `max_num_hops` is set to 0, we don't need the graph traversal.
      if max_num_hops == 0:
          if not filter_query:
              artifacts = self._store.get_artifacts_by_id(artifact_ids)
          else:
              artifacts = self._store.get_artifacts(
                  list_options=metadata_store.ListOptions(
                      filter_query=f"id IN ({artifact_ids_str}) AND ({filter_query})",
                      limit=_MAX_NUM_STARTING_NODES,
                  )
              )
          return {artifact.id: [artifact] for artifact in artifacts}
  
      options = metadata_store_pb2.LineageSubgraphQueryOptions(
          starting_artifacts=metadata_store_pb2.LineageSubgraphQueryOptions.StartingNodes(
              filter_query=f"id IN ({artifact_ids_str})"
          ),
          max_num_hops=max_num_hops,
          direction=metadata_store_pb2.LineageSubgraphQueryOptions.Direction.UPSTREAM,
      )
      field_mask_paths = (
          [_ARTIFACTS_FIELD_MASK_PATH, _EVENTS_FIELD_MASK_PATH]
          if not filter_query
          else [_EVENTS_FIELD_MASK_PATH]
      )
      lineage_graph = self._store.get_lineage_subgraph(
          query_options=options,
          field_mask_paths=field_mask_paths,
      )
  
      if not filter_query:
          artifacts_to_subgraph = metadata_resolver_utils.get_subgraphs_by_artifact_ids(
              artifact_ids,
              metadata_store_pb2.LineageSubgraphQueryOptions.Direction.UPSTREAM,
              lineage_graph,
              event_filter,
          )
          return {
              artifact_id: list(subgraph.artifacts)
              for artifact_id, subgraph in artifacts_to_subgraph.items()
          }
      else:
          artifacts_to_visited_ids = metadata_resolver_utils.get_visited_ids_by_artifact_ids(
              artifact_ids,
              metadata_store_pb2.LineageSubgraphQueryOptions.Direction.UPSTREAM,
              lineage_graph,
              event_filter,
          )
          candidate_artifact_ids = set()
          for visited_ids in artifacts_to_visited_ids.values():
              candidate_artifact_ids.update(
                  visited_ids[metadata_resolver_utils.NodeType.ARTIFACT]
              )
          artifact_ids_str = ",".join(
              str(id) for id in candidate_artifact_ids
          )
          # Send a call to metadata_store to get filtered upstream artifacts.
          artifacts = self._store.get_artifacts(
              list_options=metadata_store.ListOptions(
                  filter_query=f"id IN ({artifact_ids_str}) AND ({filter_query})"
              )
          )
          artifact_id_to_artifact = {
              artifact.id: artifact for artifact in artifacts
          }
          upstream_artifacts_dict = {}
          for (
              artifact_id,
              visited_ids,
          ) in artifacts_to_visited_ids.items():
              upstream_artifacts = [
                  artifact_id_to_artifact[id]
                  for id in visited_ids[
                      metadata_resolver_utils.NodeType.ARTIFACT
                  ]
                  if id in artifact_id_to_artifact
              ]
              if upstream_artifacts:
                  upstream_artifacts_dict[
                      artifact_id
                  ] = upstream_artifacts
          return upstream_artifacts_dict
  
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 4 for calling the MetadataResolver obj's underlying member method code with example usage: obj.get_upstream_artifacts_by_artifact_ids(artifact_ids, max_num_hops, filter_query, event_filter) and returns: dict, unknown, unknown, upstream_artifacts_dict