class_docstr: Helper class to handle metadata I/O.
  
  Not thread-safe without external synchronisation.
docstr: Gets qualified artifacts that have the right producer info.
  
  Args:
    contexts: context constraints to filter artifacts
    type_name: type constraint to filter artifacts
    producer_component_id: producer constraint to filter artifacts
    output_key: output key constraint to filter artifacts
  
  Returns:
    A list of qualified `tfx.types.Artifact`s.
docstr_tok: ["Gets","qualified","artifacts","that","have","the","right","producer","info",".","Args",":","contexts",":","context","constraints","to","filter","artifacts","type_name",":","type","constraint","to","filter","artifacts","producer_component_id",":","producer","constraint","to","filter","artifacts","output_key",":","output","key","constraint","to","filter","artifacts","Returns",":","A","list","of","qualified","`","tfx.types.Artifact","`","s","."]
code: def get_qualified_artifacts(
      self,
      contexts: List[metadata_store_pb2.Context],
      type_name: str,
      producer_component_id: Optional[str] = None,
      output_key: Optional[str] = None,
  ) -> List[_Artifact]:
      """Gets qualified artifacts that have the right producer info.
  
      Args:
        contexts: context constraints to filter artifacts
        type_name: type constraint to filter artifacts
        producer_component_id: producer constraint to filter artifacts
        output_key: output key constraint to filter artifacts
  
      Returns:
        A list of qualified `tfx.types.Artifact`s.
      """
  
      def _is_qualified_execution(execution):
          if producer_component_id is None:
              return True
          actual = self._get_legacy_producer_component_id(execution)
          return actual == producer_component_id
  
      def _is_qualified_event(event):
          return event_lib.is_valid_output_event(event, output_key)
  
      try:
          artifact_type = self.store.get_artifact_type(type_name)
          if not artifact_type:
              raise mlmd.errors.NotFoundError(
                  None,
                  None,
                  "No artifact type found for %s." % type_name,
              )
      except mlmd.errors.NotFoundError:
          return []
  
      # Gets the executions that are associated with all contexts.
      assert contexts, "Must have at least one context."
      executions_dict = {}
      valid_execution_ids = None
      for context in contexts:
          executions = self.store.get_executions_by_context(context.id)
          if valid_execution_ids is None:
              valid_execution_ids = {e.id for e in executions}
          else:
              valid_execution_ids &= {e.id for e in executions}
          executions_dict.update({e.id: e for e in executions})
  
      executions_within_context = [
          executions_dict[v] for v in valid_execution_ids
      ]
  
      # Filters the executions to match producer component id.
      qualified_producer_executions = [
          e.id
          for e in executions_within_context
          if _is_qualified_execution(e)
      ]
      # Gets the output events that have the matched output key.
      qualified_output_events = [
          ev
          for ev in self.store.get_events_by_execution_ids(
              qualified_producer_executions
          )
          if _is_qualified_event(ev)
      ]
      # Gets the candidate artifacts from output events.
      candidate_artifacts = self.store.get_artifacts_by_id(
          list(set(ev.artifact_id for ev in qualified_output_events))
      )
      # Filters the artifacts that have the right artifact type and state.
      qualified_artifacts = [
          a
          for a in candidate_artifacts
          if a.type_id == artifact_type.id
          and a.state == metadata_store_pb2.Artifact.LIVE
      ]
      return [
          artifact_utils.deserialize_artifact(artifact_type, a)
          for a in qualified_artifacts
      ]
  
imports: import collections
  import copy
  import hashlib
  import itertools
  import os
  import random
  import time
  import types
  from typing import Any, Dict, List, Optional, Set, Tuple, Type, Union
  import absl
  from tfx.dsl.io import fileio
  from tfx.orchestration import data_types
  from tfx.orchestration.portable.mlmd import event_lib
  from tfx.types import artifact
  from tfx.types import artifact_utils
  import ml_metadata
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 13 for calling the Metadata obj's underlying member method code with example usage: obj.get_qualified_artifacts(contexts, type_name, producer_component_id, output_key) and returns: unknown, unknown, event_lib, True, list