class_docstr: TFX base stub executor.
class_docstr_tok: ["TFX","base","stub","executor","."]
docstr: Initializes a BaseStubExecutor.
  
  Args:
    component_id: component id of a component associated with the stub
      executor.
    test_data_dir: The directory to test data (pipeline_recorder.py).
    context: context class for all executors.
      component_id and test_data_dir can be encoded in the context as well.
  
  Raises:
    ValueError: If the recorded pipeline data doesn't exist at test_data_dir.
docstr_tok: ["Initializes","a","BaseStubExecutor",".","Args",":","component_id",":","component","id","of","a","component","associated","with","the","stub","executor",".","test_data_dir",":","The","directory","to","test","data","(","pipeline_recorder.py",")",".","context",":","context","class","for","all","executors",".","component_id","and","test_data_dir","can","be","encoded","in","the","context","as","well",".","Raises",":","ValueError",":","If","the","recorded","pipeline","data","does","n't","exist","at","test_data_dir","."]
code: def __init__(
      self,
      component_id: Optional[str] = None,
      test_data_dir: Optional[str] = None,
      context: Optional[base_executor.BaseExecutor.Context] = None,
  ):
      """Initializes a BaseStubExecutor.
  
      Args:
        component_id: component id of a component associated with the stub
          executor.
        test_data_dir: The directory to test data (pipeline_recorder.py).
        context: context class for all executors.
          component_id and test_data_dir can be encoded in the context as well.
  
      Raises:
        ValueError: If the recorded pipeline data doesn't exist at test_data_dir.
      """
      # Fill parameters from extra_flags if empty.
      if context:
          extra_flags = context.extra_flags
          for extra_flag in extra_flags:
              if (
                  extra_flag.startswith(TEST_DATA_DIR_FLAG)
                  and test_data_dir is None
              ):
                  test_data_dir = extra_flag[
                      len(TEST_DATA_DIR_FLAG) + 1 :
                  ]  # skip '='.
              elif (
                  extra_flag.startswith(COMPONENT_ID_FLAG)
                  and component_id is None
              ):
                  component_id = extra_flag[
                      len(COMPONENT_ID_FLAG) + 1 :
                  ]  # skip '='.
  
      super().__init__(context)
      logging.info(
          "Running StubExecutor, component_id %s", component_id
      )
      self._component_id = component_id
      self._test_data_dir = test_data_dir
      if not fileio.exists(self._test_data_dir):
          raise ValueError(
              "Must record pipeline in {}".format(self._test_data_dir)
          )
  
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 1 to create a new BaseStubExecutor object from inherited base classes: base_executor with example: obj = BaseStubExecutor(component_id, test_data_dir, context)