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 resolver spec for a latest artifact resolver.
  
  Returns:
    A list of two PipelineTaskSpecs. One represents the query for latest valid
    ModelBlessing artifact. Another one represents the query for latest
    blessed Model artifact.
  Raises:
    ValueError: when desired_num_of_artifacts != 1. 1 is the only supported
      value currently.
docstr_tok: ["Builds","a","resolver","spec","for","a","latest","artifact","resolver",".","Returns",":","A","list","of","two","PipelineTaskSpecs",".","One","represents","the","query","for","latest","valid","ModelBlessing","artifact",".","Another","one","represents","the","query","for","latest","blessed","Model","artifact",".","Raises",":","ValueError",":","when","desired_num_of_artifacts","!","=","1",".","1","is","the","only","supported","value","currently","."]
code: def _build_latest_artifact_resolver(
      self,
  ) -> Dict[str, pipeline_pb2.PipelineTaskSpec]:
      """Builds a resolver spec for a latest artifact resolver.
  
      Returns:
        A list of two PipelineTaskSpecs. One represents the query for latest valid
        ModelBlessing artifact. Another one represents the query for latest
        blessed Model artifact.
      Raises:
        ValueError: when desired_num_of_artifacts != 1. 1 is the only supported
          value currently.
      """
      # Fetch the init kwargs for the resolver.
      resolver_config = self._exec_properties[resolver.RESOLVER_CONFIG]
      if (
          isinstance(resolver_config, dict)
          and resolver_config.get("desired_num_of_artifacts", 0) > 1
      ):
          raise ValueError(
              "Only desired_num_of_artifacts=1 is supported currently."
              " Got {}".format(
                  resolver_config.get("desired_num_of_artifacts")
              )
          )
  
      component_def = pipeline_pb2.ComponentSpec()
      executor_label = _EXECUTOR_LABEL_PATTERN.format(self._name)
      component_def.executor_label = executor_label
      task_spec = pipeline_pb2.PipelineTaskSpec()
      task_spec.task_info.name = self._name
  
      for name, output_channel in self._outputs.items():
          output_artifact_spec = (
              compiler_utils.build_output_artifact_spec(output_channel)
          )
          component_def.output_definitions.artifacts[name].CopyFrom(
              output_artifact_spec
          )
      for name, value in self._exec_properties.items():
          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 isinstance(value, data_types.RuntimeParameter):
              parameter_utils.attach_parameter(value)
              task_spec.inputs.parameters[
                  name
              ].component_input_parameter = value.name
          else:
              task_spec.inputs.parameters[name].CopyFrom(
                  pipeline_pb2.TaskInputsSpec.InputParameterSpec(
                      runtime_value=compiler_utils.value_converter(
                          value
                      )
                  )
              )
      self._component_defs[self._name] = component_def
      task_spec.component_ref.name = self._name
  
      artifact_queries = {}
      # Buid the artifact query for each channel in the input dict.
      for name, c in self._inputs.items():
          query_filter = (
              'schema_title="{type}" AND state={state}'
          ).format(
              type=compiler_utils.get_artifact_title(c.type),
              state=metadata_store_pb2.Artifact.State.Name(
                  metadata_store_pb2.Artifact.LIVE
              ),
          )
          # Resolver's output dict has the same set of keys as its input dict.
          artifact_queries[name] = ResolverSpec.ArtifactQuerySpec(
              filter=query_filter
          )
  
      resolver_spec = ResolverSpec(
          output_artifact_queries=artifact_queries
      )
      executor = pipeline_pb2.PipelineDeploymentConfig.ExecutorSpec()
      executor.resolver.CopyFrom(resolver_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 6 for calling the StepBuilder obj's underlying member method code with example usage: obj._build_latest_artifact_resolver() and returns: dict