class_docstr: BaseDriver is the base class of all custom drivers.
  
  This can also be used as the default driver of a component if no custom logic
  is needed.
  
  Attributes:
    _metadata_handle: An instance of Metadata.
class_docstr_tok: ["BaseDriver","is","the","base","class","of","all","custom","drivers",".","This","can","also","be","used","as","the","default","driver","of","a","component","if","no","custom","logic","is","needed",".","Attributes",":","_metadata_handle",":","An","instance","of","Metadata","."]
docstr: Handle pre-execution logic.
  
  There are four steps:
    1. Fetches input artifacts from metadata and checks whether uri exists.
    2. Registers execution.
    3. Decides whether a new execution is needed.
    4a. If (3), prepare output artifacts.
    4b. If not (3), fetch cached output artifacts.
  
  Args:
    input_dict: key -> Channel for inputs.
    output_dict: key -> Channel for outputs. Uris of the outputs are not
      assigned.
    exec_properties: Dict of other execution properties.
    driver_args: An instance of data_types.DriverArgs class.
    pipeline_info: An instance of data_types.PipelineInfo, holding pipeline
      related properties including pipeline_name, pipeline_root and run_id
    component_info: An instance of data_types.ComponentInfo, holding component
      related properties including component_type and component_id.
  
  Returns:
    data_types.ExecutionDecision object.
  
  Raises:
    RuntimeError: if any input as an empty uri.
docstr_tok: ["Handle","pre-execution","logic",".","There","are","four","steps",":","1",".","Fetches","input","artifacts","from","metadata","and","checks","whether","uri","exists",".","2",".","Registers","execution",".","3",".","Decides","whether","a","new","execution","is","needed",".","4a",".","If","(","3",")",",","prepare","output","artifacts",".","4b",".","If","not","(","3",")",",","fetch","cached","output","artifacts",".","Args",":","input_dict",":","key","-",">","Channel","for","inputs",".","output_dict",":","key","-",">","Channel","for","outputs",".","Uris","of","the","outputs","are","not","assigned",".","exec_properties",":","Dict","of","other","execution","properties",".","driver_args",":","An","instance","of","data_types.DriverArgs","class",".","pipeline_info",":","An","instance","of","data_types.PipelineInfo",",","holding","pipeline","related","properties","including","pipeline_name",",","pipeline_root","and","run_id","component_info",":","An","instance","of","data_types.ComponentInfo",",","holding","component","related","properties","including","component_type","and","component_id",".","Returns",":","data_types.ExecutionDecision","object",".","Raises",":","RuntimeError",":","if","any","input","as","an","empty","uri","."]
code: def pre_execution(
      self,
      input_dict: Dict[str, types.BaseChannel],
      output_dict: Dict[str, types.Channel],
      exec_properties: Dict[str, Any],
      driver_args: data_types.DriverArgs,
      pipeline_info: data_types.PipelineInfo,
      component_info: data_types.ComponentInfo,
  ) -> data_types.ExecutionDecision:
      """Handle pre-execution logic.
  
      There are four steps:
        1. Fetches input artifacts from metadata and checks whether uri exists.
        2. Registers execution.
        3. Decides whether a new execution is needed.
        4a. If (3), prepare output artifacts.
        4b. If not (3), fetch cached output artifacts.
  
      Args:
        input_dict: key -> Channel for inputs.
        output_dict: key -> Channel for outputs. Uris of the outputs are not
          assigned.
        exec_properties: Dict of other execution properties.
        driver_args: An instance of data_types.DriverArgs class.
        pipeline_info: An instance of data_types.PipelineInfo, holding pipeline
          related properties including pipeline_name, pipeline_root and run_id
        component_info: An instance of data_types.ComponentInfo, holding component
          related properties including component_type and component_id.
  
      Returns:
        data_types.ExecutionDecision object.
  
      Raises:
        RuntimeError: if any input as an empty uri.
      """
      # Step 1. Fetch inputs from metadata.
      exec_properties = self.resolve_exec_properties(
          exec_properties, pipeline_info, component_info
      )
      input_artifacts = self.resolve_input_artifacts(
          input_dict, exec_properties, driver_args, pipeline_info
      )
      self.verify_input_artifacts(artifacts_dict=input_artifacts)
      absl.logging.debug(
          "Resolved input artifacts are: %s", input_artifacts
      )
      # Step 2. Register execution in metadata.
      contexts = self._metadata_handle.register_pipeline_contexts_if_not_exists(
          pipeline_info
      )
      execution = self._metadata_handle.register_execution(
          input_artifacts=input_artifacts,
          exec_properties=exec_properties,
          pipeline_info=pipeline_info,
          component_info=component_info,
          contexts=contexts,
      )
      use_cached_results = False
      output_artifacts = None
  
      if driver_args.enable_cache:
          # Step 3. Decide whether a new execution is needed.
          output_artifacts = self._metadata_handle.get_cached_outputs(
              input_artifacts=input_artifacts,
              exec_properties=exec_properties,
              pipeline_info=pipeline_info,
              component_info=component_info,
          )
  
          # Check that cached output artifacts will actually be considered a cache
          # hit by downstream components
          if output_artifacts is not None:
              try:
                  artifact_utils.verify_artifacts(output_artifacts)
                  use_cached_results = True
              except RuntimeError:
                  absl.logging.debug(
                      "Cached results found but could not be verified to still exist"
                  )
  
      if use_cached_results:
          # If cache should be used, updates execution to reflect that. Note that
          # with this update, publisher should / will be skipped.
          self._metadata_handle.update_execution(
              execution=execution,
              component_info=component_info,
              output_artifacts=output_artifacts,
              execution_state=metadata.EXECUTION_STATE_CACHED,
              contexts=contexts,
          )
      else:
          absl.logging.debug(
              "Cached results not available, move on to new execution"
          )
          # Step 4a. New execution is needed. Prepare output artifacts.
          output_artifacts = self._prepare_output_artifacts(
              input_artifacts=input_artifacts,
              output_dict=output_dict,
              exec_properties=exec_properties,
              execution_id=execution.id,
              pipeline_info=pipeline_info,
              component_info=component_info,
          )
          absl.logging.debug(
              "Output artifacts skeleton for the upcoming execution are: %s",
              output_artifacts,
          )
          # Updates the execution to reflect refreshed output artifacts and
          # execution properties.
          self._metadata_handle.update_execution(
              execution=execution,
              component_info=component_info,
              output_artifacts=output_artifacts,
              exec_properties=exec_properties,
              contexts=contexts,
          )
          absl.logging.debug(
              "Execution properties for the upcoming execution are: %s",
              exec_properties,
          )
  
      # For interactive execution, update the output channel contents.
      # TODO(b/161490287): figure out the long-term behavior of Channel artifacts
      # with respect to interactive and non-interactive execution.
      if driver_args.interactive_resolution:
          for key, artifact_list in output_artifacts.items():
              channel = output_dict[key]
              channel._artifacts = (
                  artifact_list  # pylint: disable=protected-access
              )
  
      return data_types.ExecutionDecision(
          input_artifacts,
          output_artifacts,
          exec_properties,
          execution.id,
          use_cached_results,
      )
  
file_path: tfx/tfx/dsl/components/base/base_driver.py
imports: import os
  from typing import Any, Dict, List
  import absl
  from tfx import types
  from tfx.dsl.io import fileio
  from tfx.orchestration import data_types
  from tfx.orchestration import metadata
  from tfx.types import artifact_utils
  from tfx.types import channel_utils
label_desc: Use image node_id 7 for calling the BaseDriver obj's underlying member method code with example usage: obj.pre_execution(input_dict, output_dict, exec_properties, driver_args, pipeline_info, component_info) and returns: data_types