HEX
Server: Apache/2.4.65 (Ubuntu)
System: Linux ielts-store-v2 6.8.0-1036-gcp #38~22.04.1-Ubuntu SMP Thu Aug 14 01:19:18 UTC 2025 x86_64
User: root (0)
PHP: 7.2.34-54+ubuntu20.04.1+deb.sury.org+1
Disabled: pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,
Upload Files
File: //snap/google-cloud-cli/current/lib/googlecloudsdk/api_lib/dataproc/dataproc.py
# -*- coding: utf-8 -*- #
# Copyright 2015 Google LLC. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Common stateful utilities for the gcloud dataproc tool."""

from __future__ import absolute_import
from __future__ import division
from __future__ import unicode_literals

from googlecloudsdk.api_lib.util import apis
from googlecloudsdk.calliope import base
from googlecloudsdk.core import resources


class Dataproc(object):
  """Stateful utility for calling Dataproc APIs.

  While this currently could all be static. It is encapsulated in a class to
  support API version switching in future.
  """

  def __init__(self, release_track=base.ReleaseTrack.GA):
    super(Dataproc, self).__init__()
    self.release_track = release_track
    self.api_version = 'v1'
    self._client = None
    self._resources = None

  @property
  def client(self):
    if self._client is None:
      self._client = apis.GetClientInstance('dataproc', self.api_version)
    return self._client

  @property
  def messages(self):
    return apis.GetMessagesModule('dataproc', self.api_version)

  @property
  def resources(self):
    if self._resources is None:
      self._resources = resources.REGISTRY.Clone()
      self._resources.RegisterApiByName('dataproc', self.api_version)
    return self._resources

  @property
  def terminal_job_states(self):
    return [
        self.messages.JobStatus.StateValueValuesEnum.CANCELLED,
        self.messages.JobStatus.StateValueValuesEnum.DONE,
        self.messages.JobStatus.StateValueValuesEnum.ERROR,
    ]

  def GetCreateClusterRequest(self,
                              cluster,
                              project_id,
                              region,
                              request_id,
                              action_on_failed_primary_workers=None):
    """Gets the CreateClusterRequest for the appropriate api version.

    Args :
      cluster : Dataproc cluster to be created.
      project_id: The ID of the Google Cloud Platform project that the cluster
      belongs to.
      region : The Dataproc region in which to handle the request.
      request_id : A unique ID used to identify the request.
      action_on_failed_primary_workers : Supported only for v1 api.

    Raises :
      ValueError : if non-None action_on_failed_primary_workers is passed for
      v1beta2 api.

    Returns :
      DataprocProjectsRegionsClustersCreateRequest
    """
    if action_on_failed_primary_workers is None:
      return self.messages.DataprocProjectsRegionsClustersCreateRequest(
          cluster=cluster,
          projectId=project_id,
          region=region,
          requestId=request_id)

    if self.api_version == 'v1beta2':
      raise ValueError(
          'action_on_failed_primary_workers is not supported for v1beta2 api')

    return self.messages.DataprocProjectsRegionsClustersCreateRequest(
        cluster=cluster,
        projectId=project_id,
        region=region,
        requestId=request_id,
        actionOnFailedPrimaryWorkers=action_on_failed_primary_workers)

  def GetRegionsWorkflowTemplate(self, template, version=None):
    """Gets workflow template from dataproc.

    Args:
      template: workflow template resource that contains template name and id.
      version: version of the workflow template to get.

    Returns:
      WorkflowTemplate object that contains the workflow template info.

    Raises:
      ValueError: if version cannot be converted to a valid integer.
    """
    messages = self.messages
    get_request = messages.DataprocProjectsRegionsWorkflowTemplatesGetRequest(
        name=template.RelativeName())
    if version:
      get_request.version = int(version)
    return self.client.projects_regions_workflowTemplates.Get(get_request)