class_docstr: Generator implementation class for SyncPipelineTaskGenerator.
class_docstr_tok: ["Generator","implementation","class","for","SyncPipelineTaskGenerator","."]
docstr: Generates tasks for a node by freshly resolving inputs.
code: def _generate_tasks_from_resolved_inputs(
      self,
      node: node_proto_view.NodeProtoView,
  ) -> List[task_lib.Task]:
      """Generates tasks for a node by freshly resolving inputs."""
      logging.info(
          "Generating tasks from resolved inputs for node: %s",
          node.node_info.id,
      )
      result = []
      node_uid = task_lib.NodeUid.from_node(self._pipeline, node)
  
      try:
          resolved_info = task_gen_utils.generate_resolved_info(
              self._mlmd_connection_manager, node, self._pipeline
          )
          logging.info("Resolved inputs: %s", resolved_info)
      except exceptions.InputResolutionError as e:
          error_msg = (
              f"failure to resolve inputs; node uid: {node_uid}; "
              f"error: {e.__cause__ or e}"
          )
          result.append(
              self._update_node_state_to_failed_task(
                  node_uid,
                  error_code=e.grpc_code_value,
                  error_msg=error_msg,
              )
          )
          return result
  
      if not resolved_info.input_and_params:
          logging.info("Node skipped: %s", node_uid)
          result.append(
              task_lib.UpdateNodeStateTask(
                  node_uid=node_uid,
                  state=pstate.NodeState.SKIPPED,
                  status=status_lib.Status(
                      code=status_lib.Code.OK,
                      message=(
                          "Node execution skipped either due to conditional"
                          " evaluated to false or no inputs resolved. Please check"
                          " whether the output of the upstream node was generated"
                          " successfully."
                      ),
                  ),
              )
          )
          return result
  
      # Copys artifact types of the external artifacts to local db, in idempotent
      # manner. Idempotency is guaranteed by the artifact type name.
      # The external artifacts will be copies to local db when we register
      # executions. Idempotency is guaranteed by external_id.
      # TODO(b/258477751) Add more tests to test the producer/consumer pipelines.
      for input_and_params in resolved_info.input_and_params:
          for artifacts in input_and_params.input_artifacts.values():
              logging.info(
                  "Updating external artifact type: %s", artifacts
              )
              task_gen_utils.update_external_artifact_type(
                  self._mlmd_handle, artifacts
              )
  
      executions = task_gen_utils.register_executions(
          metadata_handle=self._mlmd_handle,
          execution_type=node.node_info.type,
          contexts=resolved_info.contexts,
          input_and_params=resolved_info.input_and_params,
      )
  
      result.extend(
          task_gen_utils.generate_tasks_from_one_input(
              metadata_handle=self._mlmd_handle,
              node=node,
              execution=executions[0],
              input_and_param=resolved_info.input_and_params[0],
              contexts=resolved_info.contexts,
              pipeline=self._pipeline,
              execution_node_state=pstate.NodeState.RUNNING,
          )
      )
      return result
  
imports: import collections
  import textwrap
  from typing import Callable, Dict, List, Mapping, Optional, Set
  from absl import logging
  from tfx.orchestration import node_proto_view
  from tfx.orchestration.experimental.core import mlmd_state
  from tfx.orchestration.experimental.core import pipeline_state
  from tfx.orchestration.experimental.core import service_jobs
  from tfx.orchestration.experimental.core import task
  from tfx.orchestration.experimental.core import task_gen
  from tfx.orchestration.experimental.core import task_gen_utils
  from tfx.orchestration import mlmd_connection_manager
  from tfx.orchestration.portable.input_resolution import exceptions
  from tfx.orchestration.portable.mlmd import execution_lib
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.utils import status
  from tfx.utils import topsort
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 7 for calling the _Generator obj's underlying member method code with example usage: obj._generate_tasks_from_resolved_inputs(node) and returns: result, result, result