class_docstr: TFX base stub executor.
class_docstr_tok: ["TFX","base","stub","executor","."]
docstr: Copies over recorded data to pipeline output uri.
  
  Args:
    input_dict: Input dict from input key to a list of Artifacts.
    output_dict: Output dict from output key to a list of Artifacts.
    exec_properties: A dict of execution properties.
  
  Returns:
    None
  
  Raises:
    FileNotFoundError: If the recorded test data dir doesn't exist any more.
docstr_tok: ["Copies","over","recorded","data","to","pipeline","output","uri",".","Args",":","input_dict",":","Input","dict","from","input","key","to","a","list","of","Artifacts",".","output_dict",":","Output","dict","from","output","key","to","a","list","of","Artifacts",".","exec_properties",":","A","dict","of","execution","properties",".","Returns",":","None","Raises",":","FileNotFoundError",":","If","the","recorded","test","data","dir","does","n't","exist","any","more","."]
code: def Do(
      self,
      input_dict: Dict[str, List[types.Artifact]],
      output_dict: Dict[str, List[types.Artifact]],
      exec_properties: Dict[str, Any],
  ) -> None:
      """Copies over recorded data to pipeline output uri.
  
      Args:
        input_dict: Input dict from input key to a list of Artifacts.
        output_dict: Output dict from output key to a list of Artifacts.
        exec_properties: A dict of execution properties.
  
      Returns:
        None
  
      Raises:
        FileNotFoundError: If the recorded test data dir doesn't exist any more.
      """
      for output_key, artifact_list in output_dict.items():
          for idx, artifact in enumerate(artifact_list):
              dest = artifact.uri
              src = os.path.join(
                  self._test_data_dir,
                  self._component_id,
                  output_key,
                  str(idx),
              )
              if not fileio.exists(src):
                  raise FileNotFoundError(
                      "{} does not exist".format(src)
                  )
              io_utils.copy_dir(src, dest)
              logging.info("Finished copying from %s to %s", src, dest)
  
file_path: tfx/tfx/experimental/pipeline_testing/base_stub_executor.py
imports: import os
  from typing import Any, Dict, List, Optional
  from absl import logging
  from tfx import types
  from tfx.dsl.components.base import base_executor
  from tfx.dsl.io import fileio
  from tfx.utils import io_utils
label_desc: Use image node_id 2 for calling the BaseStubExecutor obj's underlying member method code with example usage: obj.Do(input_dict, output_dict, exec_properties) without return types