class_docstr: Helper class to handle metadata I/O.
  
  Not thread-safe without external synchronisation.
docstr: Search artifacts that matches given info.
  
  Args:
    artifact_name: the name of the artifact that set by producer component.
      The name is logged both in artifacts and the events when the execution
      being published.
    pipeline_info: the information of the current pipeline
    producer_component_id: the id of the component that produces the artifact
  
  Returns:
    A list of Artifacts that matches the given info
  
  Raises:
    RuntimeError: when no matching execution is found given producer info.
docstr_tok: ["Search","artifacts","that","matches","given","info",".","Args",":","artifact_name",":","the","name","of","the","artifact","that","set","by","producer","component",".","The","name","is","logged","both","in","artifacts","and","the","events","when","the","execution","being","published",".","pipeline_info",":","the","information","of","the","current","pipeline","producer_component_id",":","the","id","of","the","component","that","produces","the","artifact","Returns",":","A","list","of","Artifacts","that","matches","the","given","info","Raises",":","RuntimeError",":","when","no","matching","execution","is","found","given","producer","info","."]
code: def search_artifacts(
      self,
      artifact_name: str,
      pipeline_info: data_types.PipelineInfo,
      producer_component_id: str,
  ) -> List[_Artifact]:
      """Search artifacts that matches given info.
  
      Args:
        artifact_name: the name of the artifact that set by producer component.
          The name is logged both in artifacts and the events when the execution
          being published.
        pipeline_info: the information of the current pipeline
        producer_component_id: the id of the component that produces the artifact
  
      Returns:
        A list of Artifacts that matches the given info
  
      Raises:
        RuntimeError: when no matching execution is found given producer info.
      """
      producer_execution = None
      matching_artifact_ids = set()
      # TODO(ruoyu): We need to revisit this when adding support for async
      # execution.
      context = self.get_pipeline_run_context(pipeline_info)
      if context is None:
          raise RuntimeError(
              "Pipeline run context for %s does not exist"
              % pipeline_info
          )
      for execution in self.store.get_executions_by_context(context.id):
          if (
              execution.properties["component_id"].string_value
              == producer_component_id
          ):
              producer_execution = execution
              break
      if not producer_execution:
          raise RuntimeError(
              "Cannot find matching execution with pipeline name %s,"
              "run id %s and component id %s"
              % (
                  pipeline_info.pipeline_name,
                  pipeline_info.run_id,
                  producer_component_id,
              )
          )
      for event in self.store.get_events_by_execution_ids(
          [producer_execution.id]
      ):
          if (
              event.type == metadata_store_pb2.Event.OUTPUT
              and event.path.steps[0].key == artifact_name
          ):
              matching_artifact_ids.add(event.artifact_id)
  
      # Get relevant artifacts along with their types.
      artifacts_by_id = self.store.get_artifacts_by_id(
          list(matching_artifact_ids)
      )
      matching_artifact_type_ids = list(
          set(a.type_id for a in artifacts_by_id)
      )
      matching_artifact_types = self.store.get_artifact_types_by_id(
          matching_artifact_type_ids
      )
      artifact_types = dict(
          zip(matching_artifact_type_ids, matching_artifact_types)
      )
  
      result_artifacts = []
      for a in artifacts_by_id:
          tfx_artifact = artifact_utils.deserialize_artifact(
              artifact_types[a.type_id], a
          )
          result_artifacts.append(tfx_artifact)
      return result_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 26 for calling the Metadata obj's underlying member method code with example usage: obj.search_artifacts(artifact_name, pipeline_info, producer_component_id) and returns: result_artifacts