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: Resolve input artifacts from metadata.
  
  Subclasses might override this function for customized artifact properties
  resolution logic. However please note that this function is supposed to be
  called in normal cases (except head of the pipeline) since it handles
  artifact info passing from upstream components.
  
  Args:
    input_dict: key -> Channel mapping for inputs generated in logical
      pipeline.
    exec_properties: Dict of other execution properties, e.g., configs.
    driver_args: An instance of data_types.DriverArgs with driver
      configuration properties.
    pipeline_info: An instance of data_types.PipelineInfo, holding pipeline
      related properties including component_type and component_id.
  
  Returns:
    Final artifacts that will be used in execution.
  
  Raises:
    ValueError: if in interactive mode, the given input channels have not been
      resolved.
docstr_tok: ["Resolve","input","artifacts","from","metadata",".","Subclasses","might","override","this","function","for","customized","artifact","properties","resolution","logic",".","However","please","note","that","this","function","is","supposed","to","be","called","in","normal","cases","(","except","head","of","the","pipeline",")","since","it","handles","artifact","info","passing","from","upstream","components",".","Args",":","input_dict",":","key","-",">","Channel","mapping","for","inputs","generated","in","logical","pipeline",".","exec_properties",":","Dict","of","other","execution","properties",",","e.g.",",","configs",".","driver_args",":","An","instance","of","data_types.DriverArgs","with","driver","configuration","properties",".","pipeline_info",":","An","instance","of","data_types.PipelineInfo",",","holding","pipeline","related","properties","including","component_type","and","component_id",".","Returns",":","Final","artifacts","that","will","be","used","in","execution",".","Raises",":","ValueError",":","if","in","interactive","mode",",","the","given","input","channels","have","not","been","resolved","."]
code: def resolve_input_artifacts(
      self,
      input_dict: Dict[str, types.BaseChannel],
      exec_properties: Dict[
          str, Any
      ],  # pylint: disable=unused-argument
      driver_args: data_types.DriverArgs,
      pipeline_info: data_types.PipelineInfo,
  ) -> Dict[str, List[types.Artifact]]:
      """Resolve input artifacts from metadata.
  
      Subclasses might override this function for customized artifact properties
      resolution logic. However please note that this function is supposed to be
      called in normal cases (except head of the pipeline) since it handles
      artifact info passing from upstream components.
  
      Args:
        input_dict: key -> Channel mapping for inputs generated in logical
          pipeline.
        exec_properties: Dict of other execution properties, e.g., configs.
        driver_args: An instance of data_types.DriverArgs with driver
          configuration properties.
        pipeline_info: An instance of data_types.PipelineInfo, holding pipeline
          related properties including component_type and component_id.
  
      Returns:
        Final artifacts that will be used in execution.
  
      Raises:
        ValueError: if in interactive mode, the given input channels have not been
          resolved.
      """
      result = {}
      for name, value in input_dict.items():
          artifacts_by_id = {}  # Deduplicate by ID.
          # TODO(b/248145891): Stop using get_individual_channels which does not
          # support all BaseChannel types. Use a common input resolution stack
          # instead.
          for input_channel in channel_utils.get_individual_channels(
              value
          ):
              if driver_args.interactive_resolution:
                  artifacts = list(input_channel.get())
                  for artifact in artifacts:
                      # Note: when not initialized, artifact.uri is '' and artifact.id is
                      # 0.
                      if not artifact.uri or not artifact.id:
                          raise ValueError(
                              f"Unresolved input channel {repr(artifact)} for input "
                              f"{repr(name)} was passed in interactive mode. When running "
                              "in interactive mode, upstream components must first be run "
                              "with `interactive_context.run(component)` before their "
                              "outputs can be used in downstream components."
                          )
                      artifacts_by_id.update(
                          {a.id: a for a in artifacts}
                      )
              else:
                  artifacts = self._metadata_handle.search_artifacts(
                      artifact_name=input_channel.output_key,
                      pipeline_info=pipeline_info,
                      producer_component_id=input_channel.producer_component_id,
                  )
                  # TODO(ccy): add this code path to interactive resolution.
                  for artifact in artifacts:
                      if isinstance(artifact, types.ValueArtifact):
                          # Resolve the content of file into value field for value
                          # artifacts.
                          _ = artifact.read()
                  artifacts_by_id.update({a.id: a for a in artifacts})
          result[name] = list(artifacts_by_id.values())
      return result
  
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 4 for calling the BaseDriver obj's underlying member method code with example usage: obj.resolve_input_artifacts(input_dict, exec_properties, driver_args, pipeline_info) and returns: result