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 the resolver spec for latest valid ModelBlessing artifact.
docstr_tok: ["Builds","the","resolver","spec","for","latest","valid","ModelBlessing","artifact","."]
code: def _build_resolver_for_latest_model_blessing(
      self, model_blessing_channel_key: str
  ) -> pipeline_pb2.PipelineTaskSpec:
      """Builds the resolver spec for latest valid ModelBlessing artifact."""
      name = "{}{}".format(self._name, _MODEL_BLESSING_RESOLVER_SUFFIX)
  
      # Component def.
      component_def = pipeline_pb2.ComponentSpec()
      executor_label = _EXECUTOR_LABEL_PATTERN.format(name)
      component_def.executor_label = executor_label
      output_artifact_spec = compiler_utils.build_output_artifact_spec(
          self._outputs[model_blessing_channel_key]
      )
      component_def.output_definitions.artifacts[
          model_blessing_channel_key
      ].CopyFrom(output_artifact_spec)
      self._component_defs[name] = component_def
  
      # Task spec.
      task_spec = pipeline_pb2.PipelineTaskSpec()
      task_spec.task_info.name = name
      task_spec.component_ref.name = name
  
      # Builds the resolver executor spec for latest valid ModelBlessing.
      executor = pipeline_pb2.PipelineDeploymentConfig.ExecutorSpec()
      artifact_queries = {}
      query_filter = (
          'schema_title="{type}" AND state={state}'
          " AND metadata.{key}.number_value={value}"
      ).format(
          type=compiler_utils.get_artifact_title(
              standard_artifacts.ModelBlessing
          ),
          state=metadata_store_pb2.Artifact.State.Name(
              metadata_store_pb2.Artifact.LIVE
          ),
          key=constants.ARTIFACT_PROPERTY_BLESSED_KEY,
          value=constants.BLESSED_VALUE,
      )
      artifact_queries[
          model_blessing_channel_key
      ] = ResolverSpec.ArtifactQuerySpec(filter=query_filter)
      executor.resolver.CopyFrom(
          ResolverSpec(output_artifact_queries=artifact_queries)
      )
      self._deployment_config.executors[executor_label].CopyFrom(
          executor
      )
  
      return 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 7 for calling the StepBuilder obj's underlying member method code with example usage: obj._build_resolver_for_latest_model_blessing(model_blessing_channel_key) and returns: task_spec