class_docstr: Kubeflow pipelines task builder.
  
  Constructs a pipeline task spec based on the TFX node information. Meanwhile,
  augments the deployment config associated with the node.
docstr: Builds a pipeline PipelineTaskSpec given the node information.
  
  Each TFX node maps one task spec and usually one component definition and
  one executor spec. (with resolver node as an exception. See explaination
  in the Returns section).
  
   - Component definition includes interfaces of a node. For example, name
  and type information of inputs/outputs/execution_properties.
   - Task spec contains the topologies around the node. For example, the
  dependency nodes, where to read the inputs and exec_properties (from another
  task, from parent component or from a constant value). The task spec has the
  name of the component definition it references. It is possible that a task
  spec references an existing component definition that's built previously.
   - Executor spec encodes how the node is actually executed. For example,
  args to start a container, or query strings for resolvers. All executor spec
  will be packed into deployment config proto.
  
  During the build, all three parts mentioned above will be updated.
  
  Returns:
    A Dict mapping from node id to PipelineTaskSpec messages corresponding to
    the node. For most of the cases, the dict contains a single element.
    The only exception is when compiling latest blessed model resolver.
    One DSL node will be split to two resolver specs to reflect the
    two-phased query execution.
  
  Raises:
    NotImplementedError: When the node being built is an InfraValidator.
docstr_tok: ["Builds","a","pipeline","PipelineTaskSpec","given","the","node","information",".","Each","TFX","node","maps","one","task","spec","and","usually","one","component","definition","and","one","executor","spec",".","(","with","resolver","node","as","an","exception",".","See","explaination","in","the","Returns","section",")",".","-","Component","definition","includes","interfaces","of","a","node",".","For","example",",","name","and","type","information","of","inputs\/outputs\/execution_properties",".","-","Task","spec","contains","the","topologies","around","the","node",".","For","example",",","the","dependency","nodes",",","where","to","read","the","inputs","and","exec_properties","(","from","another","task",",","from","parent","component","or","from","a","constant","value",")",".","The","task","spec","has","the","name","of","the","component","definition","it","references",".","It","is","possible","that","a","task","spec","references","an","existing","component","definition","that","'s","built","previously",".","-","Executor","spec","encodes","how","the","node","is","actually","executed",".","For","example",",","args","to","start","a","container",",","or","query","strings","for","resolvers",".","All","executor","spec","will","be","packed","into","deployment","config","proto",".","During","the","build",",","all","three","parts","mentioned","above","will","be","updated",".","Returns",":","A","Dict","mapping","from","node","id","to","PipelineTaskSpec","messages","corresponding","to","the","node",".","For","most","of","the","cases",",","the","dict","contains","a","single","element",".","The","only","exception","is","when","compiling","latest","blessed","model","resolver",".","One","DSL","node","will","be","split","to","two","resolver","specs","to","reflect","the","two-phased","query","execution",".","Raises",":","NotImplementedError",":","When","the","node","being","built","is","an","InfraValidator","."]
code: def build(self) -> Dict[str, pipeline_pb2.PipelineTaskSpec]:
      """Builds a pipeline PipelineTaskSpec given the node information.
  
      Each TFX node maps one task spec and usually one component definition and
      one executor spec. (with resolver node as an exception. See explaination
      in the Returns section).
  
       - Component definition includes interfaces of a node. For example, name
      and type information of inputs/outputs/execution_properties.
       - Task spec contains the topologies around the node. For example, the
      dependency nodes, where to read the inputs and exec_properties (from another
      task, from parent component or from a constant value). The task spec has the
      name of the component definition it references. It is possible that a task
      spec references an existing component definition that's built previously.
       - Executor spec encodes how the node is actually executed. For example,
      args to start a container, or query strings for resolvers. All executor spec
      will be packed into deployment config proto.
  
      During the build, all three parts mentioned above will be updated.
  
      Returns:
        A Dict mapping from node id to PipelineTaskSpec messages corresponding to
        the node. For most of the cases, the dict contains a single element.
        The only exception is when compiling latest blessed model resolver.
        One DSL node will be split to two resolver specs to reflect the
        two-phased query execution.
  
      Raises:
        NotImplementedError: When the node being built is an InfraValidator.
      """
      # 1. Resolver tasks won't have input artifacts in the API proto. First we
      #    specialcase two resolver types we support.
      if isinstance(self._node, resolver.Resolver):
          return self._build_resolver_spec()
  
      # 2. Build component spec.
      component_def = pipeline_pb2.ComponentSpec()
      task_spec = pipeline_pb2.PipelineTaskSpec()
      executor_label = _EXECUTOR_LABEL_PATTERN.format(self._name)
      component_def.executor_label = executor_label
  
      # Conditionals
      implicit_input_channels = {}
      implicit_upstream_node_ids = set()
      predicates = conditional.get_predicates(
          self._node, self._dsl_context_registry
      )
      if predicates:
          implicit_keys_map = {
              tfx_compiler_utils.implicit_channel_key(channel): key
              for key, channel in self._inputs.items()
          }
          cel_predicates = []
          for predicate in predicates:
              for channel in channel_utils.get_dependent_channels(
                  predicate
              ):
                  implicit_key = (
                      tfx_compiler_utils.implicit_channel_key(channel)
                  )
                  if implicit_key not in implicit_keys_map:
                      # Store this channel and add it to the node inputs later.
                      implicit_input_channels[implicit_key] = channel
                      # Store the producer node and add it to the upstream nodes later.
                      implicit_upstream_node_ids.add(
                          channel.producer_component_id
                      )
              placeholder_pb = (
                  channel_utils.encode_placeholder_with_channels(
                      predicate,
                      tfx_compiler_utils.build_channel_to_key_fn(
                          implicit_keys_map
                      ),
                  )
              )
              cel_predicates.append(
                  compiler_utils.placeholder_to_cel(placeholder_pb)
              )
          task_spec.trigger_policy.condition = " && ".join(
              cel_predicates
          )
  
      # Inputs
      for name, input_channel in itertools.chain(
          self._inputs.items(), implicit_input_channels.items()
      ):
          input_artifact_spec = (
              compiler_utils.build_input_artifact_spec(input_channel)
          )
          component_def.input_definitions.artifacts[name].CopyFrom(
              input_artifact_spec
          )
      # Outputs
      for name, output_channel in self._outputs.items():
          # Currently, we're working under the assumption that for tasks
          # (those generated by BaseComponent), each channel contains a single
          # artifact.
          output_artifact_spec = (
              compiler_utils.build_output_artifact_spec(output_channel)
          )
          component_def.output_definitions.artifacts[name].CopyFrom(
              output_artifact_spec
          )
  
          # If output artifact is passed to downstream components as param, emit
          # an output param.
          key = (self._node.id, name)
          if (
              self._dynamic_exec_properties
              and key in self._dynamic_exec_properties
          ):
              output_parameter_spec = (
                  compiler_utils.build_output_parameter_spec(
                      self._dynamic_exec_properties[key]
                  )
              )
              component_def.output_definitions.parameters[
                  name
              ].CopyFrom(output_parameter_spec)
  
      # Exec properties
      for name, value in self._exec_properties.items():
          # value can be None for unprovided optional exec properties.
          if value is None:
              continue
  
          parameter_type_spec = (
              compiler_utils.build_parameter_type_spec(value)
          )
          component_def.input_definitions.parameters[name].CopyFrom(
              parameter_type_spec
          )
      if self._name not in self._component_defs:
          self._component_defs[self._name] = component_def
      else:
          raise ValueError(
              f"Found duplicate component ids {self._name} while "
              "building component definitions."
          )
  
      # 3. Build task spec.
      task_spec.task_info.name = self._name
      dependency_ids = sorted(
          {node.id for node in self._node.upstream_nodes}
          | implicit_upstream_node_ids
      )
  
      for name, input_channel in itertools.chain(
          self._inputs.items(), implicit_input_channels.items()
      ):
          # TODO(b/169573945): Add support for vertex if requested.
          if not isinstance(input_channel, Channel):
              raise TypeError("Only single Channel is supported.")
          if self._is_exit_handler:
              logging.error(
                  "exit handler component doesn't take input artifact, "
                  "the input will be ignored."
              )
              continue
          # If the redirecting map is provided (usually for latest blessed model
          # resolver, we'll need to redirect accordingly. Also, the upstream node
          # list will be updated and replaced by the new producer id.
          producer_id = input_channel.producer_component_id
          output_key = input_channel.output_key
          for k, v in self._channel_redirect_map.items():
              if k[0] == producer_id and producer_id in dependency_ids:
                  dependency_ids.remove(producer_id)
                  dependency_ids.append(v[0])
          producer_id = self._channel_redirect_map.get(
              (producer_id, output_key), (producer_id, output_key)
          )[0]
          output_key = self._channel_redirect_map.get(
              (producer_id, output_key), (producer_id, output_key)
          )[1]
  
          input_artifact_spec = (
              pipeline_pb2.TaskInputsSpec.InputArtifactSpec()
          )
          input_artifact_spec.task_output_artifact.producer_task = (
              producer_id
          )
          input_artifact_spec.task_output_artifact.output_artifact_key = (
              output_key
          )
          task_spec.inputs.artifacts[name].CopyFrom(input_artifact_spec)
  
      for name, value in self._exec_properties.items():
          if value is None:
              continue
          if isinstance(value, placeholder.Placeholder):
              try:
                  # This unwraps channel.future()[0].value and disallows any other
                  # placeholder expressions.
                  channel = (
                      channel_utils.unwrap_simple_channel_placeholder(
                          value
                      )
                  )
              except ValueError as e:
                  raise ValueError(
                      f"Invalid placeholder for exec prop {name}"
                  ) from e
              task_spec.inputs.parameters[name].CopyFrom(
                  pipeline_pb2.TaskInputsSpec.InputParameterSpec(
                      task_output_parameter=pipeline_pb2.TaskInputsSpec.InputParameterSpec.TaskOutputParameterSpec(
                          producer_task=channel.producer_component_id
                          + "_task",
                          output_parameter_key=channel.output_key,
                      )
                  )
              )
          elif isinstance(value, data_types.RuntimeParameter):
              parameter_utils.attach_parameter(value)
              task_spec.inputs.parameters[
                  name
              ].component_input_parameter = value.name
          elif isinstance(value, decorators.FinalStatusStr):
              if not self._is_exit_handler:
                  logging.error(
                      "FinalStatusStr type is only allowed to use in exit"
                      " handler. The parameter is ignored."
                  )
              else:
                  task_spec.inputs.parameters[
                      name
                  ].task_final_status.producer_task = utils.TFX_DAG_NAME
          else:
              task_spec.inputs.parameters[name].CopyFrom(
                  pipeline_pb2.TaskInputsSpec.InputParameterSpec(
                      runtime_value=compiler_utils.value_converter(
                          value
                      )
                  )
              )
      task_spec.component_ref.name = self._name
      dependency_ids = sorted(dependency_ids)
      for dependency in dependency_ids:
          task_spec.dependent_tasks.append(dependency)
  
      if self._enable_cache:
          task_spec.caching_options.CopyFrom(
              pipeline_pb2.PipelineTaskSpec.CachingOptions(
                  enable_cache=self._enable_cache
              )
          )
  
      if self._is_exit_handler:
          task_spec.trigger_policy.strategy = (
              pipeline_pb2.PipelineTaskSpec.TriggerPolicy.ALL_UPSTREAM_TASKS_COMPLETED
          )
          task_spec.dependent_tasks.append(utils.TFX_DAG_NAME)
      # 4. Build the executor body for other common tasks.
      executor = pipeline_pb2.PipelineDeploymentConfig.ExecutorSpec()
      if isinstance(self._node, importer.Importer):
          executor.importer.CopyFrom(self._build_importer_spec())
      elif isinstance(self._node, components.FileBasedExampleGen):
          executor.container.CopyFrom(
              self._build_file_based_example_gen_spec()
          )
      elif isinstance(self._node, (components.InfraValidator)):
          raise NotImplementedError(
              'The component type "{}" is not supported'.format(
                  type(self._node)
              )
          )
      else:
          executor.container.CopyFrom(self._build_container_spec())
      self._deployment_config.executors[executor_label].CopyFrom(
          executor
      )
      return {self._name: task_spec}
  
imports: import itertools
  from typing import Any, Dict, List, Optional, Tuple, cast
  from absl import logging
  from kfp.pipeline_spec import pipeline_spec_pb2
  from tfx import components
  from tfx.components.evaluator import constants
  from tfx.dsl.compiler import compiler_utils
  from tfx.dsl.component.experimental import executor_specs
  from tfx.dsl.component.experimental import placeholders
  from tfx.dsl.components.base import base_component
  from tfx.dsl.components.base import base_node
  from tfx.dsl.components.base import executor_spec
  from tfx.dsl.components.common import importer
  from tfx.dsl.components.common import resolver
  from tfx.dsl.context_managers import dsl_context_registry
  from tfx.dsl.experimental.conditionals import conditional
  from tfx.dsl.input_resolution.strategies import latest_artifact_strategy
  from tfx.dsl.input_resolution.strategies import latest_blessed_model_strategy
  from tfx.dsl.placeholder import placeholder
  from tfx.orchestration import data_types
  from tfx.orchestration.kubeflow import decorators
  from tfx.orchestration.kubeflow import utils
  from tfx.orchestration.kubeflow.v2 import compiler_utils
  from tfx.orchestration.kubeflow.v2 import parameter_utils
  from tfx.types import channel_utils
  from tfx.types import standard_artifacts
  from tfx.types.channel import Channel
  from tfx.utils import deprecation_utils
  from tfx.utils import name_utils
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 2 for calling the StepBuilder obj's underlying member method code with example usage: obj.build() and returns: dict, self