class_docstr: The handler for the system Importer node.
class_docstr_tok: ["The","handler","for","the","system","Importer","node","."]
docstr: Runs Importer specific logic.
  
  Args:
    mlmd_connection: ML metadata connection.
    pipeline_node: The specification of the node that this launcher lauches.
    pipeline_info: The information of the pipeline that this node runs in.
    pipeline_runtime_spec: The runtime information of the pipeline that this
      node runs in.
  
  Returns:
    The execution of the run.
docstr_tok: ["Runs","Importer","specific","logic",".","Args",":","mlmd_connection",":","ML","metadata","connection",".","pipeline_node",":","The","specification","of","the","node","that","this","launcher","lauches",".","pipeline_info",":","The","information","of","the","pipeline","that","this","node","runs","in",".","pipeline_runtime_spec",":","The","runtime","information","of","the","pipeline","that","this","node","runs","in",".","Returns",":","The","execution","of","the","run","."]
code: def run(
      self,
      mlmd_connection: metadata.Metadata,
      pipeline_node: pipeline_pb2.PipelineNode,
      pipeline_info: pipeline_pb2.PipelineInfo,
      pipeline_runtime_spec: pipeline_pb2.PipelineRuntimeSpec,
  ) -> data_types.ExecutionInfo:
      """Runs Importer specific logic.
  
      Args:
        mlmd_connection: ML metadata connection.
        pipeline_node: The specification of the node that this launcher lauches.
        pipeline_info: The information of the pipeline that this node runs in.
        pipeline_runtime_spec: The runtime information of the pipeline that this
          node runs in.
  
      Returns:
        The execution of the run.
      """
      logging.info("Running as an importer node.")
      with mlmd_connection as m:
          # 1.Prepares all contexts.
          contexts = context_lib.prepare_contexts(
              metadata_handle=m, node_contexts=pipeline_node.contexts
          )
  
          # 2. Resolves execution properties, please note that importers has no
          # input.
          exec_properties = data_types_utils.build_parsed_value_dict(
              inputs_utils.resolve_parameters_with_schema(
                  node_parameters=pipeline_node.parameters
              )
          )
  
          # 3. Registers execution in metadata.
          execution = execution_publish_utils.register_execution(
              metadata_handle=m,
              execution_type=pipeline_node.node_info.type,
              contexts=contexts,
              exec_properties=exec_properties,
          )
  
          # 4. Generate output artifacts to represent the imported artifacts.
          output_key = cast(
              str, exec_properties[importer.OUTPUT_KEY_KEY]
          )
          output_spec = pipeline_node.outputs.outputs[output_key]
          properties = self._extract_proto_map(
              output_spec.artifact_spec.additional_properties
          )
          custom_properties = self._extract_proto_map(
              output_spec.artifact_spec.additional_custom_properties
          )
          output_artifact_class = types.Artifact(
              output_spec.artifact_spec.type
          ).type
          output_artifacts = importer.generate_output_dict(
              metadata_handle=m,
              uri=cast(str, exec_properties[importer.SOURCE_URI_KEY]),
              properties=properties,
              custom_properties=custom_properties,
              reimport=bool(
                  exec_properties[importer.REIMPORT_OPTION_KEY]
              ),
              output_artifact_class=output_artifact_class,
              mlmd_artifact_type=output_spec.artifact_spec.type,
              output_key=output_key,
          )
  
          result = data_types.ExecutionInfo(
              execution_id=execution.id,
              input_dict={},
              output_dict=output_artifacts,
              exec_properties=exec_properties,
              pipeline_node=pipeline_node,
              pipeline_info=pipeline_info,
          )
  
          # 5. Publish the output artifacts. If artifacts are reimported, the
          # execution is published as CACHED. Otherwise it is published as COMPLETE.
          if _is_artifact_reimported(output_artifacts, output_key):
              execution_publish_utils.publish_cached_executions(
                  metadata_handle=m,
                  contexts=contexts,
                  executions=[execution],
                  output_artifacts_maps=[output_artifacts],
              )
  
          else:
              execution_publish_utils.publish_succeeded_execution(
                  metadata_handle=m,
                  execution_id=execution.id,
                  contexts=contexts,
                  output_artifacts=output_artifacts,
              )
  
          return result
  
file_path: tfx/tfx/orchestration/portable/importer_node_handler.py
imports: from typing import Any, Dict, List, cast
  from absl import logging
  from tfx import types
  from tfx.dsl.components.common import importer
  from tfx.orchestration import data_types_utils
  from tfx.orchestration import metadata
  from tfx.orchestration.portable import data_types
  from tfx.orchestration.portable import execution_publish_utils
  from tfx.orchestration.portable import inputs_utils
  from tfx.orchestration.portable import system_node_handler
  from tfx.orchestration.portable.mlmd import context_lib
  from tfx.proto.orchestration import pipeline_pb2
label_desc: Use image node_id 2 for calling the ImporterNodeHandler obj's underlying member method code with example usage: obj.run(mlmd_connection, pipeline_node, pipeline_info, pipeline_runtime_spec) and returns: result