beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From al...@apache.org
Subject [03/22] beam git commit: Rename google_cloud_dataflow and google_cloud_platform
Date Thu, 23 Feb 2017 01:22:58 GMT
http://git-wip-us.apache.org/repos/asf/beam/blob/59ad58ac/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/apiclient.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/apiclient.py b/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/apiclient.py
deleted file mode 100644
index 98473ca..0000000
--- a/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/apiclient.py
+++ /dev/null
@@ -1,726 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You 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.
-#
-
-"""Dataflow client utility functions."""
-
-import codecs
-import getpass
-import json
-import logging
-import os
-import re
-import time
-from StringIO import StringIO
-from datetime import datetime
-
-from apitools.base.py import encoding
-from apitools.base.py import exceptions
-
-from apache_beam import utils
-from apache_beam.internal.auth import get_service_credentials
-from apache_beam.internal.google_cloud_platform.json_value import to_json_value
-from apache_beam.io.google_cloud_platform.internal.clients import storage
-from apache_beam.runners.google_cloud_dataflow.internal.clients import dataflow
-from apache_beam.transforms import cy_combiners
-from apache_beam.transforms.display import DisplayData
-from apache_beam.utils import dependency
-from apache_beam.utils import retry
-from apache_beam.utils.dependency import get_required_container_version
-from apache_beam.utils.dependency import get_sdk_name_and_version
-from apache_beam.utils.names import PropertyNames
-from apache_beam.utils.pipeline_options import DebugOptions
-from apache_beam.utils.pipeline_options import GoogleCloudOptions
-from apache_beam.utils.pipeline_options import StandardOptions
-from apache_beam.utils.pipeline_options import WorkerOptions
-
-
-class Step(object):
-  """Wrapper for a dataflow Step protobuf."""
-
-  def __init__(self, step_kind, step_name, additional_properties=None):
-    self.step_kind = step_kind
-    self.step_name = step_name
-    self.proto = dataflow.Step(kind=step_kind, name=step_name)
-    self.proto.properties = {}
-    self._additional_properties = []
-
-    if additional_properties is not None:
-      for (n, v, t) in additional_properties:
-        self.add_property(n, v, t)
-
-  def add_property(self, name, value, with_type=False):
-    self._additional_properties.append((name, value, with_type))
-    self.proto.properties.additionalProperties.append(
-        dataflow.Step.PropertiesValue.AdditionalProperty(
-            key=name, value=to_json_value(value, with_type=with_type)))
-
-  def _get_outputs(self):
-    """Returns a list of all output labels for a step."""
-    outputs = []
-    for p in self.proto.properties.additionalProperties:
-      if p.key == PropertyNames.OUTPUT_INFO:
-        for entry in p.value.array_value.entries:
-          for entry_prop in entry.object_value.properties:
-            if entry_prop.key == PropertyNames.OUTPUT_NAME:
-              outputs.append(entry_prop.value.string_value)
-    return outputs
-
-  def __reduce__(self):
-    """Reduce hook for pickling the Step class more easily."""
-    return (Step, (self.step_kind, self.step_name, self._additional_properties))
-
-  def get_output(self, tag=None):
-    """Returns name if it is one of the outputs or first output if name is None.
-
-    Args:
-      tag: tag of the output as a string or None if we want to get the
-        name of the first output.
-
-    Returns:
-      The name of the output associated with the tag or the first output
-      if tag was None.
-
-    Raises:
-      ValueError: if the tag does not exist within outputs.
-    """
-    outputs = self._get_outputs()
-    if tag is None:
-      return outputs[0]
-    else:
-      name = '%s_%s' % (PropertyNames.OUT, tag)
-      if name not in outputs:
-        raise ValueError(
-            'Cannot find named output: %s in %s.' % (name, outputs))
-      return name
-
-
-class Environment(object):
-  """Wrapper for a dataflow Environment protobuf."""
-
-  def __init__(self, packages, options, environment_version):
-    self.standard_options = options.view_as(StandardOptions)
-    self.google_cloud_options = options.view_as(GoogleCloudOptions)
-    self.worker_options = options.view_as(WorkerOptions)
-    self.debug_options = options.view_as(DebugOptions)
-    self.proto = dataflow.Environment()
-    self.proto.clusterManagerApiService = GoogleCloudOptions.COMPUTE_API_SERVICE
-    self.proto.dataset = '{}/cloud_dataflow'.format(
-        GoogleCloudOptions.BIGQUERY_API_SERVICE)
-    self.proto.tempStoragePrefix = (
-        self.google_cloud_options.temp_location.replace(
-            'gs:/',
-            GoogleCloudOptions.STORAGE_API_SERVICE))
-    # User agent information.
-    self.proto.userAgent = dataflow.Environment.UserAgentValue()
-    self.local = 'localhost' in self.google_cloud_options.dataflow_endpoint
-
-    if self.google_cloud_options.service_account_email:
-      self.proto.serviceAccountEmail = (
-          self.google_cloud_options.service_account_email)
-
-    sdk_name, version_string = get_sdk_name_and_version()
-
-    self.proto.userAgent.additionalProperties.extend([
-        dataflow.Environment.UserAgentValue.AdditionalProperty(
-            key='name',
-            value=to_json_value(sdk_name)),
-        dataflow.Environment.UserAgentValue.AdditionalProperty(
-            key='version', value=to_json_value(version_string))])
-    # Version information.
-    self.proto.version = dataflow.Environment.VersionValue()
-    if self.standard_options.streaming:
-      job_type = 'PYTHON_STREAMING'
-    else:
-      job_type = 'PYTHON_BATCH'
-    self.proto.version.additionalProperties.extend([
-        dataflow.Environment.VersionValue.AdditionalProperty(
-            key='job_type',
-            value=to_json_value(job_type)),
-        dataflow.Environment.VersionValue.AdditionalProperty(
-            key='major', value=to_json_value(environment_version))])
-    # Experiments
-    if self.debug_options.experiments:
-      for experiment in self.debug_options.experiments:
-        self.proto.experiments.append(experiment)
-    # Worker pool(s) information.
-    package_descriptors = []
-    for package in packages:
-      package_descriptors.append(
-          dataflow.Package(
-              location='%s/%s' % (
-                  self.google_cloud_options.staging_location.replace(
-                      'gs:/', GoogleCloudOptions.STORAGE_API_SERVICE),
-                  package),
-              name=package))
-
-    pool = dataflow.WorkerPool(
-        kind='local' if self.local else 'harness',
-        packages=package_descriptors,
-        taskrunnerSettings=dataflow.TaskRunnerSettings(
-            parallelWorkerSettings=dataflow.WorkerSettings(
-                baseUrl=GoogleCloudOptions.DATAFLOW_ENDPOINT,
-                servicePath=self.google_cloud_options.dataflow_endpoint)))
-    pool.autoscalingSettings = dataflow.AutoscalingSettings()
-    # Set worker pool options received through command line.
-    if self.worker_options.num_workers:
-      pool.numWorkers = self.worker_options.num_workers
-    if self.worker_options.max_num_workers:
-      pool.autoscalingSettings.maxNumWorkers = (
-          self.worker_options.max_num_workers)
-    if self.worker_options.autoscaling_algorithm:
-      values_enum = dataflow.AutoscalingSettings.AlgorithmValueValuesEnum
-      pool.autoscalingSettings.algorithm = {
-          'NONE': values_enum.AUTOSCALING_ALGORITHM_NONE,
-          'THROUGHPUT_BASED': values_enum.AUTOSCALING_ALGORITHM_BASIC,
-      }.get(self.worker_options.autoscaling_algorithm)
-    if self.worker_options.machine_type:
-      pool.machineType = self.worker_options.machine_type
-    if self.worker_options.disk_size_gb:
-      pool.diskSizeGb = self.worker_options.disk_size_gb
-    if self.worker_options.disk_type:
-      pool.diskType = self.worker_options.disk_type
-    if self.worker_options.zone:
-      pool.zone = self.worker_options.zone
-    if self.worker_options.network:
-      pool.network = self.worker_options.network
-    if self.worker_options.worker_harness_container_image:
-      pool.workerHarnessContainerImage = (
-          self.worker_options.worker_harness_container_image)
-    else:
-      # Default to using the worker harness container image for the current SDK
-      # version.
-      pool.workerHarnessContainerImage = (
-          'dataflow.gcr.io/v1beta3/python:%s' %
-          get_required_container_version())
-    if self.worker_options.use_public_ips is not None:
-      if self.worker_options.use_public_ips:
-        pool.ipConfiguration = (
-            dataflow.WorkerPool
-            .IpConfigurationValueValuesEnum.WORKER_IP_PUBLIC)
-      else:
-        pool.ipConfiguration = (
-            dataflow.WorkerPool
-            .IpConfigurationValueValuesEnum.WORKER_IP_PRIVATE)
-
-    if self.standard_options.streaming:
-      # Use separate data disk for streaming.
-      disk = dataflow.Disk()
-      if self.local:
-        disk.diskType = 'local'
-      # TODO(ccy): allow customization of disk.
-      pool.dataDisks.append(disk)
-    self.proto.workerPools.append(pool)
-
-    sdk_pipeline_options = options.get_all_options()
-    if sdk_pipeline_options:
-      self.proto.sdkPipelineOptions = (
-          dataflow.Environment.SdkPipelineOptionsValue())
-
-      options_dict = {k: v
-                      for k, v in sdk_pipeline_options.iteritems()
-                      if v is not None}
-      self.proto.sdkPipelineOptions.additionalProperties.append(
-          dataflow.Environment.SdkPipelineOptionsValue.AdditionalProperty(
-              key='options', value=to_json_value(options_dict)))
-
-      dd = DisplayData.create_from_options(options)
-      items = [item.get_dict() for item in dd.items]
-      self.proto.sdkPipelineOptions.additionalProperties.append(
-          dataflow.Environment.SdkPipelineOptionsValue.AdditionalProperty(
-              key='display_data', value=to_json_value(items)))
-
-
-class Job(object):
-  """Wrapper for a dataflow Job protobuf."""
-
-  def __str__(self):
-    def encode_shortstrings(input_buffer, errors='strict'):
-      """Encoder (from Unicode) that suppresses long base64 strings."""
-      original_len = len(input_buffer)
-      if original_len > 150:
-        if self.base64_str_re.match(input_buffer):
-          input_buffer = '<string of %d bytes>' % original_len
-          input_buffer = input_buffer.encode('ascii', errors=errors)
-        else:
-          matched = self.coder_str_re.match(input_buffer)
-          if matched:
-            input_buffer = '%s<string of %d bytes>' % (
-                matched.group(1), matched.end(2) - matched.start(2))
-            input_buffer = input_buffer.encode('ascii', errors=errors)
-      return input_buffer, original_len
-
-    def decode_shortstrings(input_buffer, errors='strict'):
-      """Decoder (to Unicode) that suppresses long base64 strings."""
-      shortened, length = encode_shortstrings(input_buffer, errors)
-      return unicode(shortened), length
-
-    def shortstrings_registerer(encoding_name):
-      if encoding_name == 'shortstrings':
-        return codecs.CodecInfo(name='shortstrings',
-                                encode=encode_shortstrings,
-                                decode=decode_shortstrings)
-      return None
-
-    codecs.register(shortstrings_registerer)
-
-    # Use json "dump string" method to get readable formatting;
-    # further modify it to not output too-long strings, aimed at the
-    # 10,000+ character hex-encoded "serialized_fn" values.
-    return json.dumps(
-        json.loads(encoding.MessageToJson(self.proto), encoding='shortstrings'),
-        indent=2, sort_keys=True)
-
-  @staticmethod
-  def default_job_name(job_name):
-    if job_name is None:
-      user_name = getpass.getuser().lower()
-      date_component = datetime.utcnow().strftime('%m%d%H%M%S-%f')
-      app_name = 'beamapp'
-      job_name = '{}-{}-{}'.format(app_name, user_name, date_component)
-    return job_name
-
-  def __init__(self, options):
-    self.options = options
-    self.google_cloud_options = options.view_as(GoogleCloudOptions)
-    if not self.google_cloud_options.job_name:
-      self.google_cloud_options.job_name = self.default_job_name(
-          self.google_cloud_options.job_name)
-
-    required_google_cloud_options = ['project', 'job_name', 'temp_location']
-    missing = [
-        option for option in required_google_cloud_options
-        if not getattr(self.google_cloud_options, option)]
-    if missing:
-      raise ValueError(
-          'Missing required configuration parameters: %s' % missing)
-
-    if not self.google_cloud_options.staging_location:
-      logging.info('Defaulting to the temp_location as staging_location: %s',
-                   self.google_cloud_options.temp_location)
-      (self.google_cloud_options
-       .staging_location) = self.google_cloud_options.temp_location
-
-    # Make the staging and temp locations job name and time specific. This is
-    # needed to avoid clashes between job submissions using the same staging
-    # area or team members using same job names. This method is not entirely
-    # foolproof since two job submissions with same name can happen at exactly
-    # the same time. However the window is extremely small given that
-    # time.time() has at least microseconds granularity. We add the suffix only
-    # for GCS staging locations where the potential for such clashes is high.
-    if self.google_cloud_options.staging_location.startswith('gs://'):
-      path_suffix = '%s.%f' % (self.google_cloud_options.job_name, time.time())
-      self.google_cloud_options.staging_location = utils.path.join(
-          self.google_cloud_options.staging_location, path_suffix)
-      self.google_cloud_options.temp_location = utils.path.join(
-          self.google_cloud_options.temp_location, path_suffix)
-    self.proto = dataflow.Job(name=self.google_cloud_options.job_name)
-    if self.options.view_as(StandardOptions).streaming:
-      self.proto.type = dataflow.Job.TypeValueValuesEnum.JOB_TYPE_STREAMING
-    else:
-      self.proto.type = dataflow.Job.TypeValueValuesEnum.JOB_TYPE_BATCH
-    self.base64_str_re = re.compile(r'^[A-Za-z0-9+/]*=*$')
-    self.coder_str_re = re.compile(r'^([A-Za-z]+\$)([A-Za-z0-9+/]*=*)$')
-
-  def json(self):
-    return encoding.MessageToJson(self.proto)
-
-  def __reduce__(self):
-    """Reduce hook for pickling the Job class more easily."""
-    return (Job, (self.options,))
-
-
-class DataflowApplicationClient(object):
-  """A Dataflow API client used by application code to create and query jobs."""
-
-  def __init__(self, options, environment_version):
-    """Initializes a Dataflow API client object."""
-    self.standard_options = options.view_as(StandardOptions)
-    self.google_cloud_options = options.view_as(GoogleCloudOptions)
-    self.environment_version = environment_version
-    if self.google_cloud_options.no_auth:
-      credentials = None
-    else:
-      credentials = get_service_credentials()
-    self._client = dataflow.DataflowV1b3(
-        url=self.google_cloud_options.dataflow_endpoint,
-        credentials=credentials,
-        get_credentials=(not self.google_cloud_options.no_auth))
-    self._storage_client = storage.StorageV1(
-        url='https://www.googleapis.com/storage/v1',
-        credentials=credentials,
-        get_credentials=(not self.google_cloud_options.no_auth))
-
-  # TODO(silviuc): Refactor so that retry logic can be applied.
-  @retry.no_retries  # Using no_retries marks this as an integration point.
-  def _gcs_file_copy(self, from_path, to_path):
-    to_folder, to_name = os.path.split(to_path)
-    with open(from_path, 'rb') as f:
-      self.stage_file(to_folder, to_name, f)
-
-  def stage_file(self, gcs_or_local_path, file_name, stream,
-                 mime_type='application/octet-stream'):
-    """Stages a file at a GCS or local path with stream-supplied contents."""
-    if not gcs_or_local_path.startswith('gs://'):
-      local_path = os.path.join(gcs_or_local_path, file_name)
-      logging.info('Staging file locally to %s', local_path)
-      with open(local_path, 'wb') as f:
-        f.write(stream.read())
-      return
-    gcs_location = gcs_or_local_path + '/' + file_name
-    bucket, name = gcs_location[5:].split('/', 1)
-
-    request = storage.StorageObjectsInsertRequest(
-        bucket=bucket, name=name)
-    logging.info('Starting GCS upload to %s...', gcs_location)
-    upload = storage.Upload(stream, mime_type)
-    try:
-      response = self._storage_client.objects.Insert(request, upload=upload)
-    except exceptions.HttpError as e:
-      reportable_errors = {
-          403: 'access denied',
-          404: 'bucket not found',
-      }
-      if e.status_code in reportable_errors:
-        raise IOError(('Could not upload to GCS path %s: %s. Please verify '
-                       'that credentials are valid and that you have write '
-                       'access to the specified path. Stale credentials can be '
-                       'refreshed by executing "gcloud auth login".') %
-                      (gcs_or_local_path, reportable_errors[e.status_code]))
-      raise
-    logging.info('Completed GCS upload to %s', gcs_location)
-    return response
-
-  # TODO(silviuc): Refactor so that retry logic can be applied.
-  @retry.no_retries  # Using no_retries marks this as an integration point.
-  def create_job(self, job):
-    """Creates job description. May stage and/or submit for remote execution."""
-    self.create_job_description(job)
-
-    # Stage and submit the job when necessary
-    dataflow_job_file = job.options.view_as(DebugOptions).dataflow_job_file
-    template_location = (
-        job.options.view_as(GoogleCloudOptions).template_location)
-
-    job_location = template_location or dataflow_job_file
-    if job_location:
-      gcs_or_local_path = os.path.dirname(job_location)
-      file_name = os.path.basename(job_location)
-      self.stage_file(gcs_or_local_path, file_name, StringIO(job.json()))
-
-    if not template_location:
-      return self.submit_job_description(job)
-    else:
-      return None
-
-  def create_job_description(self, job):
-    """Creates a job described by the workflow proto."""
-    resources = dependency.stage_job_resources(
-        job.options, file_copy=self._gcs_file_copy)
-    job.proto.environment = Environment(
-        packages=resources, options=job.options,
-        environment_version=self.environment_version).proto
-    # TODO(silviuc): Remove the debug logging eventually.
-    logging.info('JOB: %s', job)
-
-  def submit_job_description(self, job):
-    """Creates and excutes a job request."""
-    request = dataflow.DataflowProjectsJobsCreateRequest()
-    request.projectId = self.google_cloud_options.project
-    request.job = job.proto
-
-    try:
-      response = self._client.projects_jobs.Create(request)
-    except exceptions.BadStatusCodeError as e:
-      logging.error('HTTP status %d trying to create job'
-                    ' at dataflow service endpoint %s',
-                    e.response.status,
-                    self.google_cloud_options.dataflow_endpoint)
-      logging.fatal('details of server error: %s', e)
-      raise
-    logging.info('Create job: %s', response)
-    # The response is a Job proto with the id for the new job.
-    logging.info('Created job with id: [%s]', response.id)
-    logging.info(
-        'To access the Dataflow monitoring console, please navigate to '
-        'https://console.developers.google.com/project/%s/dataflow/job/%s',
-        self.google_cloud_options.project, response.id)
-
-    return response
-
-  @retry.with_exponential_backoff()  # Using retry defaults from utils/retry.py
-  def modify_job_state(self, job_id, new_state):
-    """Modify the run state of the job.
-
-    Args:
-      job_id: The id of the job.
-      new_state: A string representing the new desired state. It could be set to
-      either 'JOB_STATE_DONE', 'JOB_STATE_CANCELLED' or 'JOB_STATE_DRAINING'.
-
-    Returns:
-      True if the job was modified successfully.
-    """
-    if new_state == 'JOB_STATE_DONE':
-      new_state = dataflow.Job.RequestedStateValueValuesEnum.JOB_STATE_DONE
-    elif new_state == 'JOB_STATE_CANCELLED':
-      new_state = dataflow.Job.RequestedStateValueValuesEnum.JOB_STATE_CANCELLED
-    elif new_state == 'JOB_STATE_DRAINING':
-      new_state = dataflow.Job.RequestedStateValueValuesEnum.JOB_STATE_DRAINING
-    else:
-      # Other states could only be set by the service.
-      return False
-
-    request = dataflow.DataflowProjectsJobsUpdateRequest()
-    request.jobId = job_id
-    request.projectId = self.google_cloud_options.project
-    request.job = dataflow.Job(requestedState=new_state)
-
-    self._client.projects_jobs.Update(request)
-    return True
-
-  @retry.with_exponential_backoff()  # Using retry defaults from utils/retry.py
-  def get_job(self, job_id):
-    """Gets the job status for a submitted job.
-
-    Args:
-      job_id: A string representing the job_id for the workflow as returned
-        by the a create_job() request.
-
-    Returns:
-      A Job proto. See below for interesting fields.
-
-    The Job proto returned from a get_job() request contains some interesting
-    fields:
-      currentState: An object representing the current state of the job. The
-        string representation of the object (str() result) has the following
-        possible values: JOB_STATE_UNKNONW, JOB_STATE_STOPPED,
-        JOB_STATE_RUNNING, JOB_STATE_DONE, JOB_STATE_FAILED,
-        JOB_STATE_CANCELLED.
-      createTime: UTC time when the job was created
-        (e.g. '2015-03-10T00:01:53.074Z')
-      currentStateTime: UTC time for the current state of the job.
-    """
-    request = dataflow.DataflowProjectsJobsGetRequest()
-    request.jobId = job_id
-    request.projectId = self.google_cloud_options.project
-    response = self._client.projects_jobs.Get(request)
-    return response
-
-  @retry.with_exponential_backoff()  # Using retry defaults from utils/retry.py
-  def list_messages(
-      self, job_id, start_time=None, end_time=None, page_token=None,
-      minimum_importance=None):
-    """List messages associated with the execution of a job.
-
-    Args:
-      job_id: A string representing the job_id for the workflow as returned
-        by the a create_job() request.
-      start_time: If specified, only messages generated after the start time
-        will be returned, otherwise all messages since job started will be
-        returned. The value is a string representing UTC time
-        (e.g., '2015-08-18T21:03:50.644Z')
-      end_time: If specified, only messages generated before the end time
-        will be returned, otherwise all messages up to current time will be
-        returned. The value is a string representing UTC time
-        (e.g., '2015-08-18T21:03:50.644Z')
-      page_token: A string to be used as next page token if the list call
-        returned paginated results.
-      minimum_importance: Filter for messages based on importance. The possible
-        string values in increasing order of importance are: JOB_MESSAGE_DEBUG,
-        JOB_MESSAGE_DETAILED, JOB_MESSAGE_BASIC, JOB_MESSAGE_WARNING,
-        JOB_MESSAGE_ERROR. For example, a filter set on warning will allow only
-        warnings and errors and exclude all others.
-
-    Returns:
-      A tuple consisting of a list of JobMessage instances and a
-      next page token string.
-
-    Raises:
-      RuntimeError: if an unexpected value for the message_importance argument
-        is used.
-
-    The JobMessage objects returned by the call contain the following  fields:
-      id: A unique string identifier for the message.
-      time: A string representing the UTC time of the message
-        (e.g., '2015-08-18T21:03:50.644Z')
-      messageImportance: An enumeration value for the message importance. The
-        value if converted to string will have the following possible values:
-        JOB_MESSAGE_DEBUG, JOB_MESSAGE_DETAILED, JOB_MESSAGE_BASIC,
-        JOB_MESSAGE_WARNING, JOB_MESSAGE_ERROR.
-     messageText: A message string.
-    """
-    request = dataflow.DataflowProjectsJobsMessagesListRequest(
-        jobId=job_id, projectId=self.google_cloud_options.project)
-    if page_token is not None:
-      request.pageToken = page_token
-    if start_time is not None:
-      request.startTime = start_time
-    if end_time is not None:
-      request.endTime = end_time
-    if minimum_importance is not None:
-      if minimum_importance == 'JOB_MESSAGE_DEBUG':
-        request.minimumImportance = (
-            dataflow.DataflowProjectsJobsMessagesListRequest
-            .MinimumImportanceValueValuesEnum
-            .JOB_MESSAGE_DEBUG)
-      elif minimum_importance == 'JOB_MESSAGE_DETAILED':
-        request.minimumImportance = (
-            dataflow.DataflowProjectsJobsMessagesListRequest
-            .MinimumImportanceValueValuesEnum
-            .JOB_MESSAGE_DETAILED)
-      elif minimum_importance == 'JOB_MESSAGE_BASIC':
-        request.minimumImportance = (
-            dataflow.DataflowProjectsJobsMessagesListRequest
-            .MinimumImportanceValueValuesEnum
-            .JOB_MESSAGE_BASIC)
-      elif minimum_importance == 'JOB_MESSAGE_WARNING':
-        request.minimumImportance = (
-            dataflow.DataflowProjectsJobsMessagesListRequest
-            .MinimumImportanceValueValuesEnum
-            .JOB_MESSAGE_WARNING)
-      elif minimum_importance == 'JOB_MESSAGE_ERROR':
-        request.minimumImportance = (
-            dataflow.DataflowProjectsJobsMessagesListRequest
-            .MinimumImportanceValueValuesEnum
-            .JOB_MESSAGE_ERROR)
-      else:
-        raise RuntimeError(
-            'Unexpected value for minimum_importance argument: %r',
-            minimum_importance)
-    response = self._client.projects_jobs_messages.List(request)
-    return response.jobMessages, response.nextPageToken
-
-
-class MetricUpdateTranslators(object):
-  """Translators between accumulators and dataflow metric updates."""
-
-  @staticmethod
-  def translate_boolean(accumulator, metric_update_proto):
-    metric_update_proto.boolean = accumulator.value
-
-  @staticmethod
-  def translate_scalar_mean_int(accumulator, metric_update_proto):
-    if accumulator.count:
-      metric_update_proto.integerMean = dataflow.IntegerMean()
-      metric_update_proto.integerMean.sum = to_split_int(accumulator.sum)
-      metric_update_proto.integerMean.count = to_split_int(accumulator.count)
-    else:
-      metric_update_proto.nameAndKind.kind = None
-
-  @staticmethod
-  def translate_scalar_mean_float(accumulator, metric_update_proto):
-    if accumulator.count:
-      metric_update_proto.floatingPointMean = dataflow.FloatingPointMean()
-      metric_update_proto.floatingPointMean.sum = accumulator.sum
-      metric_update_proto.floatingPointMean.count = to_split_int(
-          accumulator.count)
-    else:
-      metric_update_proto.nameAndKind.kind = None
-
-  @staticmethod
-  def translate_scalar_counter_int(accumulator, metric_update_proto):
-    metric_update_proto.integer = to_split_int(accumulator.value)
-
-  @staticmethod
-  def translate_scalar_counter_float(accumulator, metric_update_proto):
-    metric_update_proto.floatingPoint = accumulator.value
-
-
-def to_split_int(n):
-  res = dataflow.SplitInt64()
-  res.lowBits = n & 0xffffffff
-  res.highBits = n >> 32
-  return res
-
-
-def translate_distribution(distribution_update, metric_update_proto):
-  """Translate metrics DistributionUpdate to dataflow distribution update."""
-  dist_update_proto = dataflow.DistributionUpdate()
-  dist_update_proto.min = to_split_int(distribution_update.min)
-  dist_update_proto.max = to_split_int(distribution_update.max)
-  dist_update_proto.count = to_split_int(distribution_update.count)
-  dist_update_proto.sum = to_split_int(distribution_update.sum)
-  metric_update_proto.distribution = dist_update_proto
-
-
-def translate_value(value, metric_update_proto):
-  metric_update_proto.integer = to_split_int(value)
-
-
-def translate_scalar(accumulator, metric_update):
-  metric_update.scalar = to_json_value(accumulator.value, with_type=True)
-
-
-def translate_mean(accumulator, metric_update):
-  if accumulator.count:
-    metric_update.meanSum = to_json_value(accumulator.sum, with_type=True)
-    metric_update.meanCount = to_json_value(accumulator.count, with_type=True)
-  else:
-    # A denominator of 0 will raise an error in the service.
-    # What it means is we have nothing to report yet, so don't.
-    metric_update.kind = None
-
-
-# To enable a counter on the service, add it to this dictionary.
-metric_translations = {
-    cy_combiners.CountCombineFn: ('sum', translate_scalar),
-    cy_combiners.SumInt64Fn: ('sum', translate_scalar),
-    cy_combiners.MinInt64Fn: ('min', translate_scalar),
-    cy_combiners.MaxInt64Fn: ('max', translate_scalar),
-    cy_combiners.MeanInt64Fn: ('mean', translate_mean),
-    cy_combiners.SumFloatFn: ('sum', translate_scalar),
-    cy_combiners.MinFloatFn: ('min', translate_scalar),
-    cy_combiners.MaxFloatFn: ('max', translate_scalar),
-    cy_combiners.MeanFloatFn: ('mean', translate_mean),
-    cy_combiners.AllCombineFn: ('and', translate_scalar),
-    cy_combiners.AnyCombineFn: ('or', translate_scalar),
-}
-
-counter_translations = {
-    cy_combiners.CountCombineFn: (
-        dataflow.NameAndKind.KindValueValuesEnum.SUM,
-        MetricUpdateTranslators.translate_scalar_counter_int),
-    cy_combiners.SumInt64Fn: (
-        dataflow.NameAndKind.KindValueValuesEnum.SUM,
-        MetricUpdateTranslators.translate_scalar_counter_int),
-    cy_combiners.MinInt64Fn: (
-        dataflow.NameAndKind.KindValueValuesEnum.MIN,
-        MetricUpdateTranslators.translate_scalar_counter_int),
-    cy_combiners.MaxInt64Fn: (
-        dataflow.NameAndKind.KindValueValuesEnum.MAX,
-        MetricUpdateTranslators.translate_scalar_counter_int),
-    cy_combiners.MeanInt64Fn: (
-        dataflow.NameAndKind.KindValueValuesEnum.MEAN,
-        MetricUpdateTranslators.translate_scalar_mean_int),
-    cy_combiners.SumFloatFn: (
-        dataflow.NameAndKind.KindValueValuesEnum.SUM,
-        MetricUpdateTranslators.translate_scalar_counter_float),
-    cy_combiners.MinFloatFn: (
-        dataflow.NameAndKind.KindValueValuesEnum.MIN,
-        MetricUpdateTranslators.translate_scalar_counter_float),
-    cy_combiners.MaxFloatFn: (
-        dataflow.NameAndKind.KindValueValuesEnum.MAX,
-        MetricUpdateTranslators.translate_scalar_counter_float),
-    cy_combiners.MeanFloatFn: (
-        dataflow.NameAndKind.KindValueValuesEnum.MEAN,
-        MetricUpdateTranslators.translate_scalar_mean_float),
-    cy_combiners.AllCombineFn: (
-        dataflow.NameAndKind.KindValueValuesEnum.AND,
-        MetricUpdateTranslators.translate_boolean),
-    cy_combiners.AnyCombineFn: (
-        dataflow.NameAndKind.KindValueValuesEnum.OR,
-        MetricUpdateTranslators.translate_boolean),
-}

http://git-wip-us.apache.org/repos/asf/beam/blob/59ad58ac/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/apiclient_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/apiclient_test.py b/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/apiclient_test.py
deleted file mode 100644
index 2c53e37..0000000
--- a/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/apiclient_test.py
+++ /dev/null
@@ -1,96 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You 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.
-#
-"""Unit tests for the apiclient module."""
-
-import unittest
-
-from mock import Mock
-
-from apache_beam.metrics.cells import DistributionData
-from apache_beam.utils.pipeline_options import PipelineOptions
-
-from apache_beam.runners.google_cloud_dataflow.dataflow_runner import DataflowRunner
-from apache_beam.runners.google_cloud_dataflow.internal.clients import dataflow
-
-# Protect against environments where apitools library is not available.
-# pylint: disable=wrong-import-order, wrong-import-position
-try:
-  from apache_beam.runners.google_cloud_dataflow.internal import apiclient
-except ImportError:
-  apiclient = None
-# pylint: enable=wrong-import-order, wrong-import-position
-
-
-@unittest.skipIf(apiclient is None, 'GCP dependencies are not installed')
-class UtilTest(unittest.TestCase):
-
-  @unittest.skip("Enable once BEAM-1080 is fixed.")
-  def test_create_application_client(self):
-    pipeline_options = PipelineOptions()
-    apiclient.DataflowApplicationClient(
-        pipeline_options,
-        DataflowRunner.BATCH_ENVIRONMENT_MAJOR_VERSION)
-
-  def test_default_job_name(self):
-    job_name = apiclient.Job.default_job_name(None)
-    regexp = 'beamapp-.*-[0-9]{10}-[0-9]{6}'
-    self.assertRegexpMatches(job_name, regexp)
-
-  def test_split_int(self):
-    number = 12345
-    split_number = apiclient.to_split_int(number)
-    self.assertEqual((split_number.lowBits, split_number.highBits),
-                     (number, 0))
-    shift_number = number << 32
-    split_number = apiclient.to_split_int(shift_number)
-    self.assertEqual((split_number.lowBits, split_number.highBits),
-                     (0, number))
-
-  def test_translate_distribution(self):
-    metric_update = dataflow.CounterUpdate()
-    distribution_update = DistributionData(16, 2, 1, 15)
-    apiclient.translate_distribution(distribution_update, metric_update)
-    self.assertEqual(metric_update.distribution.min.lowBits,
-                     distribution_update.min)
-    self.assertEqual(metric_update.distribution.max.lowBits,
-                     distribution_update.max)
-    self.assertEqual(metric_update.distribution.sum.lowBits,
-                     distribution_update.sum)
-    self.assertEqual(metric_update.distribution.count.lowBits,
-                     distribution_update.count)
-
-  def test_translate_means(self):
-    metric_update = dataflow.CounterUpdate()
-    accumulator = Mock()
-    accumulator.sum = 16
-    accumulator.count = 2
-    apiclient.MetricUpdateTranslators.translate_scalar_mean_int(accumulator,
-                                                                metric_update)
-    self.assertEqual(metric_update.integerMean.sum.lowBits, accumulator.sum)
-    self.assertEqual(metric_update.integerMean.count.lowBits, accumulator.count)
-
-    accumulator.sum = 16.0
-    accumulator.count = 2
-    apiclient.MetricUpdateTranslators.translate_scalar_mean_float(accumulator,
-                                                                  metric_update)
-    self.assertEqual(metric_update.floatingPointMean.sum, accumulator.sum)
-    self.assertEqual(
-        metric_update.floatingPointMean.count.lowBits, accumulator.count)
-
-
-if __name__ == '__main__':
-  unittest.main()

http://git-wip-us.apache.org/repos/asf/beam/blob/59ad58ac/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/clients/__init__.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/clients/__init__.py b/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/clients/__init__.py
deleted file mode 100644
index cce3aca..0000000
--- a/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/clients/__init__.py
+++ /dev/null
@@ -1,16 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You 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.
-#

http://git-wip-us.apache.org/repos/asf/beam/blob/59ad58ac/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/clients/dataflow/__init__.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/clients/dataflow/__init__.py b/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/clients/dataflow/__init__.py
deleted file mode 100644
index 1399f21..0000000
--- a/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/clients/dataflow/__init__.py
+++ /dev/null
@@ -1,33 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You 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 imports for generated dataflow client library."""
-# pylint:disable=wildcard-import
-
-import pkgutil
-
-# Protect against environments where apitools library is not available.
-# pylint: disable=wrong-import-order, wrong-import-position
-try:
-  from apitools.base.py import *
-  from apache_beam.runners.google_cloud_dataflow.internal.clients.dataflow.dataflow_v1b3_messages import *
-  from apache_beam.runners.google_cloud_dataflow.internal.clients.dataflow.dataflow_v1b3_client import *
-except ImportError:
-  pass
-# pylint: enable=wrong-import-order, wrong-import-position
-
-__path__ = pkgutil.extend_path(__path__, __name__)

http://git-wip-us.apache.org/repos/asf/beam/blob/59ad58ac/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/clients/dataflow/dataflow_v1b3_client.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/clients/dataflow/dataflow_v1b3_client.py b/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/clients/dataflow/dataflow_v1b3_client.py
deleted file mode 100644
index 4d3d525..0000000
--- a/sdks/python/apache_beam/runners/google_cloud_dataflow/internal/clients/dataflow/dataflow_v1b3_client.py
+++ /dev/null
@@ -1,684 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You 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.
-#
-
-"""Generated client library for dataflow version v1b3."""
-# NOTE: This file is autogenerated and should not be edited by hand.
-from apitools.base.py import base_api
-
-from apache_beam.runners.google_cloud_dataflow.internal.clients.dataflow import dataflow_v1b3_messages as messages
-
-
-class DataflowV1b3(base_api.BaseApiClient):
-  """Generated client library for service dataflow version v1b3."""
-
-  MESSAGES_MODULE = messages
-  BASE_URL = u'https://dataflow.googleapis.com/'
-
-  _PACKAGE = u'dataflow'
-  _SCOPES = [u'https://www.googleapis.com/auth/cloud-platform', u'https://www.googleapis.com/auth/userinfo.email']
-  _VERSION = u'v1b3'
-  _CLIENT_ID = '1042881264118.apps.googleusercontent.com'
-  _CLIENT_SECRET = 'x_Tw5K8nnjoRAqULM9PFAC2b'
-  _USER_AGENT = 'x_Tw5K8nnjoRAqULM9PFAC2b'
-  _CLIENT_CLASS_NAME = u'DataflowV1b3'
-  _URL_VERSION = u'v1b3'
-  _API_KEY = None
-
-  def __init__(self, url='', credentials=None,
-               get_credentials=True, http=None, model=None,
-               log_request=False, log_response=False,
-               credentials_args=None, default_global_params=None,
-               additional_http_headers=None):
-    """Create a new dataflow handle."""
-    url = url or self.BASE_URL
-    super(DataflowV1b3, self).__init__(
-        url, credentials=credentials,
-        get_credentials=get_credentials, http=http, model=model,
-        log_request=log_request, log_response=log_response,
-        credentials_args=credentials_args,
-        default_global_params=default_global_params,
-        additional_http_headers=additional_http_headers)
-    self.projects_jobs_debug = self.ProjectsJobsDebugService(self)
-    self.projects_jobs_messages = self.ProjectsJobsMessagesService(self)
-    self.projects_jobs_workItems = self.ProjectsJobsWorkItemsService(self)
-    self.projects_jobs = self.ProjectsJobsService(self)
-    self.projects_locations_jobs_messages = self.ProjectsLocationsJobsMessagesService(self)
-    self.projects_locations_jobs_workItems = self.ProjectsLocationsJobsWorkItemsService(self)
-    self.projects_locations_jobs = self.ProjectsLocationsJobsService(self)
-    self.projects_locations = self.ProjectsLocationsService(self)
-    self.projects_templates = self.ProjectsTemplatesService(self)
-    self.projects = self.ProjectsService(self)
-
-  class ProjectsJobsDebugService(base_api.BaseApiService):
-    """Service class for the projects_jobs_debug resource."""
-
-    _NAME = u'projects_jobs_debug'
-
-    def __init__(self, client):
-      super(DataflowV1b3.ProjectsJobsDebugService, self).__init__(client)
-      self._upload_configs = {
-          }
-
-    def GetConfig(self, request, global_params=None):
-      """Get encoded debug configuration for component. Not cacheable.
-
-      Args:
-        request: (DataflowProjectsJobsDebugGetConfigRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (GetDebugConfigResponse) The response message.
-      """
-      config = self.GetMethodConfig('GetConfig')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    GetConfig.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'POST',
-        method_id=u'dataflow.projects.jobs.debug.getConfig',
-        ordered_params=[u'projectId', u'jobId'],
-        path_params=[u'jobId', u'projectId'],
-        query_params=[],
-        relative_path=u'v1b3/projects/{projectId}/jobs/{jobId}/debug/getConfig',
-        request_field=u'getDebugConfigRequest',
-        request_type_name=u'DataflowProjectsJobsDebugGetConfigRequest',
-        response_type_name=u'GetDebugConfigResponse',
-        supports_download=False,
-    )
-
-    def SendCapture(self, request, global_params=None):
-      """Send encoded debug capture data for component.
-
-      Args:
-        request: (DataflowProjectsJobsDebugSendCaptureRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (SendDebugCaptureResponse) The response message.
-      """
-      config = self.GetMethodConfig('SendCapture')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    SendCapture.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'POST',
-        method_id=u'dataflow.projects.jobs.debug.sendCapture',
-        ordered_params=[u'projectId', u'jobId'],
-        path_params=[u'jobId', u'projectId'],
-        query_params=[],
-        relative_path=u'v1b3/projects/{projectId}/jobs/{jobId}/debug/sendCapture',
-        request_field=u'sendDebugCaptureRequest',
-        request_type_name=u'DataflowProjectsJobsDebugSendCaptureRequest',
-        response_type_name=u'SendDebugCaptureResponse',
-        supports_download=False,
-    )
-
-  class ProjectsJobsMessagesService(base_api.BaseApiService):
-    """Service class for the projects_jobs_messages resource."""
-
-    _NAME = u'projects_jobs_messages'
-
-    def __init__(self, client):
-      super(DataflowV1b3.ProjectsJobsMessagesService, self).__init__(client)
-      self._upload_configs = {
-          }
-
-    def List(self, request, global_params=None):
-      """Request the job status.
-
-      Args:
-        request: (DataflowProjectsJobsMessagesListRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (ListJobMessagesResponse) The response message.
-      """
-      config = self.GetMethodConfig('List')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    List.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'GET',
-        method_id=u'dataflow.projects.jobs.messages.list',
-        ordered_params=[u'projectId', u'jobId'],
-        path_params=[u'jobId', u'projectId'],
-        query_params=[u'endTime', u'location', u'minimumImportance', u'pageSize', u'pageToken', u'startTime'],
-        relative_path=u'v1b3/projects/{projectId}/jobs/{jobId}/messages',
-        request_field='',
-        request_type_name=u'DataflowProjectsJobsMessagesListRequest',
-        response_type_name=u'ListJobMessagesResponse',
-        supports_download=False,
-    )
-
-  class ProjectsJobsWorkItemsService(base_api.BaseApiService):
-    """Service class for the projects_jobs_workItems resource."""
-
-    _NAME = u'projects_jobs_workItems'
-
-    def __init__(self, client):
-      super(DataflowV1b3.ProjectsJobsWorkItemsService, self).__init__(client)
-      self._upload_configs = {
-          }
-
-    def Lease(self, request, global_params=None):
-      """Leases a dataflow WorkItem to run.
-
-      Args:
-        request: (DataflowProjectsJobsWorkItemsLeaseRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (LeaseWorkItemResponse) The response message.
-      """
-      config = self.GetMethodConfig('Lease')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    Lease.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'POST',
-        method_id=u'dataflow.projects.jobs.workItems.lease',
-        ordered_params=[u'projectId', u'jobId'],
-        path_params=[u'jobId', u'projectId'],
-        query_params=[],
-        relative_path=u'v1b3/projects/{projectId}/jobs/{jobId}/workItems:lease',
-        request_field=u'leaseWorkItemRequest',
-        request_type_name=u'DataflowProjectsJobsWorkItemsLeaseRequest',
-        response_type_name=u'LeaseWorkItemResponse',
-        supports_download=False,
-    )
-
-    def ReportStatus(self, request, global_params=None):
-      """Reports the status of dataflow WorkItems leased by a worker.
-
-      Args:
-        request: (DataflowProjectsJobsWorkItemsReportStatusRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (ReportWorkItemStatusResponse) The response message.
-      """
-      config = self.GetMethodConfig('ReportStatus')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    ReportStatus.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'POST',
-        method_id=u'dataflow.projects.jobs.workItems.reportStatus',
-        ordered_params=[u'projectId', u'jobId'],
-        path_params=[u'jobId', u'projectId'],
-        query_params=[],
-        relative_path=u'v1b3/projects/{projectId}/jobs/{jobId}/workItems:reportStatus',
-        request_field=u'reportWorkItemStatusRequest',
-        request_type_name=u'DataflowProjectsJobsWorkItemsReportStatusRequest',
-        response_type_name=u'ReportWorkItemStatusResponse',
-        supports_download=False,
-    )
-
-  class ProjectsJobsService(base_api.BaseApiService):
-    """Service class for the projects_jobs resource."""
-
-    _NAME = u'projects_jobs'
-
-    def __init__(self, client):
-      super(DataflowV1b3.ProjectsJobsService, self).__init__(client)
-      self._upload_configs = {
-          }
-
-    def Create(self, request, global_params=None):
-      """Creates a Cloud Dataflow job.
-
-      Args:
-        request: (DataflowProjectsJobsCreateRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (Job) The response message.
-      """
-      config = self.GetMethodConfig('Create')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    Create.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'POST',
-        method_id=u'dataflow.projects.jobs.create',
-        ordered_params=[u'projectId'],
-        path_params=[u'projectId'],
-        query_params=[u'location', u'replaceJobId', u'view'],
-        relative_path=u'v1b3/projects/{projectId}/jobs',
-        request_field=u'job',
-        request_type_name=u'DataflowProjectsJobsCreateRequest',
-        response_type_name=u'Job',
-        supports_download=False,
-    )
-
-    def Get(self, request, global_params=None):
-      """Gets the state of the specified Cloud Dataflow job.
-
-      Args:
-        request: (DataflowProjectsJobsGetRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (Job) The response message.
-      """
-      config = self.GetMethodConfig('Get')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    Get.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'GET',
-        method_id=u'dataflow.projects.jobs.get',
-        ordered_params=[u'projectId', u'jobId'],
-        path_params=[u'jobId', u'projectId'],
-        query_params=[u'location', u'view'],
-        relative_path=u'v1b3/projects/{projectId}/jobs/{jobId}',
-        request_field='',
-        request_type_name=u'DataflowProjectsJobsGetRequest',
-        response_type_name=u'Job',
-        supports_download=False,
-    )
-
-    def GetMetrics(self, request, global_params=None):
-      """Request the job status.
-
-      Args:
-        request: (DataflowProjectsJobsGetMetricsRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (JobMetrics) The response message.
-      """
-      config = self.GetMethodConfig('GetMetrics')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    GetMetrics.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'GET',
-        method_id=u'dataflow.projects.jobs.getMetrics',
-        ordered_params=[u'projectId', u'jobId'],
-        path_params=[u'jobId', u'projectId'],
-        query_params=[u'location', u'startTime'],
-        relative_path=u'v1b3/projects/{projectId}/jobs/{jobId}/metrics',
-        request_field='',
-        request_type_name=u'DataflowProjectsJobsGetMetricsRequest',
-        response_type_name=u'JobMetrics',
-        supports_download=False,
-    )
-
-    def List(self, request, global_params=None):
-      """List the jobs of a project.
-
-      Args:
-        request: (DataflowProjectsJobsListRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (ListJobsResponse) The response message.
-      """
-      config = self.GetMethodConfig('List')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    List.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'GET',
-        method_id=u'dataflow.projects.jobs.list',
-        ordered_params=[u'projectId'],
-        path_params=[u'projectId'],
-        query_params=[u'filter', u'location', u'pageSize', u'pageToken', u'view'],
-        relative_path=u'v1b3/projects/{projectId}/jobs',
-        request_field='',
-        request_type_name=u'DataflowProjectsJobsListRequest',
-        response_type_name=u'ListJobsResponse',
-        supports_download=False,
-    )
-
-    def Update(self, request, global_params=None):
-      """Updates the state of an existing Cloud Dataflow job.
-
-      Args:
-        request: (DataflowProjectsJobsUpdateRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (Job) The response message.
-      """
-      config = self.GetMethodConfig('Update')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    Update.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'PUT',
-        method_id=u'dataflow.projects.jobs.update',
-        ordered_params=[u'projectId', u'jobId'],
-        path_params=[u'jobId', u'projectId'],
-        query_params=[u'location'],
-        relative_path=u'v1b3/projects/{projectId}/jobs/{jobId}',
-        request_field=u'job',
-        request_type_name=u'DataflowProjectsJobsUpdateRequest',
-        response_type_name=u'Job',
-        supports_download=False,
-    )
-
-  class ProjectsLocationsJobsMessagesService(base_api.BaseApiService):
-    """Service class for the projects_locations_jobs_messages resource."""
-
-    _NAME = u'projects_locations_jobs_messages'
-
-    def __init__(self, client):
-      super(DataflowV1b3.ProjectsLocationsJobsMessagesService, self).__init__(client)
-      self._upload_configs = {
-          }
-
-    def List(self, request, global_params=None):
-      """Request the job status.
-
-      Args:
-        request: (DataflowProjectsLocationsJobsMessagesListRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (ListJobMessagesResponse) The response message.
-      """
-      config = self.GetMethodConfig('List')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    List.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'GET',
-        method_id=u'dataflow.projects.locations.jobs.messages.list',
-        ordered_params=[u'projectId', u'location', u'jobId'],
-        path_params=[u'jobId', u'location', u'projectId'],
-        query_params=[u'endTime', u'minimumImportance', u'pageSize', u'pageToken', u'startTime'],
-        relative_path=u'v1b3/projects/{projectId}/locations/{location}/jobs/{jobId}/messages',
-        request_field='',
-        request_type_name=u'DataflowProjectsLocationsJobsMessagesListRequest',
-        response_type_name=u'ListJobMessagesResponse',
-        supports_download=False,
-    )
-
-  class ProjectsLocationsJobsWorkItemsService(base_api.BaseApiService):
-    """Service class for the projects_locations_jobs_workItems resource."""
-
-    _NAME = u'projects_locations_jobs_workItems'
-
-    def __init__(self, client):
-      super(DataflowV1b3.ProjectsLocationsJobsWorkItemsService, self).__init__(client)
-      self._upload_configs = {
-          }
-
-    def Lease(self, request, global_params=None):
-      """Leases a dataflow WorkItem to run.
-
-      Args:
-        request: (DataflowProjectsLocationsJobsWorkItemsLeaseRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (LeaseWorkItemResponse) The response message.
-      """
-      config = self.GetMethodConfig('Lease')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    Lease.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'POST',
-        method_id=u'dataflow.projects.locations.jobs.workItems.lease',
-        ordered_params=[u'projectId', u'location', u'jobId'],
-        path_params=[u'jobId', u'location', u'projectId'],
-        query_params=[],
-        relative_path=u'v1b3/projects/{projectId}/locations/{location}/jobs/{jobId}/workItems:lease',
-        request_field=u'leaseWorkItemRequest',
-        request_type_name=u'DataflowProjectsLocationsJobsWorkItemsLeaseRequest',
-        response_type_name=u'LeaseWorkItemResponse',
-        supports_download=False,
-    )
-
-    def ReportStatus(self, request, global_params=None):
-      """Reports the status of dataflow WorkItems leased by a worker.
-
-      Args:
-        request: (DataflowProjectsLocationsJobsWorkItemsReportStatusRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (ReportWorkItemStatusResponse) The response message.
-      """
-      config = self.GetMethodConfig('ReportStatus')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    ReportStatus.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'POST',
-        method_id=u'dataflow.projects.locations.jobs.workItems.reportStatus',
-        ordered_params=[u'projectId', u'location', u'jobId'],
-        path_params=[u'jobId', u'location', u'projectId'],
-        query_params=[],
-        relative_path=u'v1b3/projects/{projectId}/locations/{location}/jobs/{jobId}/workItems:reportStatus',
-        request_field=u'reportWorkItemStatusRequest',
-        request_type_name=u'DataflowProjectsLocationsJobsWorkItemsReportStatusRequest',
-        response_type_name=u'ReportWorkItemStatusResponse',
-        supports_download=False,
-    )
-
-  class ProjectsLocationsJobsService(base_api.BaseApiService):
-    """Service class for the projects_locations_jobs resource."""
-
-    _NAME = u'projects_locations_jobs'
-
-    def __init__(self, client):
-      super(DataflowV1b3.ProjectsLocationsJobsService, self).__init__(client)
-      self._upload_configs = {
-          }
-
-    def Create(self, request, global_params=None):
-      """Creates a Cloud Dataflow job.
-
-      Args:
-        request: (DataflowProjectsLocationsJobsCreateRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (Job) The response message.
-      """
-      config = self.GetMethodConfig('Create')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    Create.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'POST',
-        method_id=u'dataflow.projects.locations.jobs.create',
-        ordered_params=[u'projectId', u'location'],
-        path_params=[u'location', u'projectId'],
-        query_params=[u'replaceJobId', u'view'],
-        relative_path=u'v1b3/projects/{projectId}/locations/{location}/jobs',
-        request_field=u'job',
-        request_type_name=u'DataflowProjectsLocationsJobsCreateRequest',
-        response_type_name=u'Job',
-        supports_download=False,
-    )
-
-    def Get(self, request, global_params=None):
-      """Gets the state of the specified Cloud Dataflow job.
-
-      Args:
-        request: (DataflowProjectsLocationsJobsGetRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (Job) The response message.
-      """
-      config = self.GetMethodConfig('Get')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    Get.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'GET',
-        method_id=u'dataflow.projects.locations.jobs.get',
-        ordered_params=[u'projectId', u'location', u'jobId'],
-        path_params=[u'jobId', u'location', u'projectId'],
-        query_params=[u'view'],
-        relative_path=u'v1b3/projects/{projectId}/locations/{location}/jobs/{jobId}',
-        request_field='',
-        request_type_name=u'DataflowProjectsLocationsJobsGetRequest',
-        response_type_name=u'Job',
-        supports_download=False,
-    )
-
-    def GetMetrics(self, request, global_params=None):
-      """Request the job status.
-
-      Args:
-        request: (DataflowProjectsLocationsJobsGetMetricsRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (JobMetrics) The response message.
-      """
-      config = self.GetMethodConfig('GetMetrics')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    GetMetrics.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'GET',
-        method_id=u'dataflow.projects.locations.jobs.getMetrics',
-        ordered_params=[u'projectId', u'location', u'jobId'],
-        path_params=[u'jobId', u'location', u'projectId'],
-        query_params=[u'startTime'],
-        relative_path=u'v1b3/projects/{projectId}/locations/{location}/jobs/{jobId}/metrics',
-        request_field='',
-        request_type_name=u'DataflowProjectsLocationsJobsGetMetricsRequest',
-        response_type_name=u'JobMetrics',
-        supports_download=False,
-    )
-
-    def List(self, request, global_params=None):
-      """List the jobs of a project.
-
-      Args:
-        request: (DataflowProjectsLocationsJobsListRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (ListJobsResponse) The response message.
-      """
-      config = self.GetMethodConfig('List')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    List.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'GET',
-        method_id=u'dataflow.projects.locations.jobs.list',
-        ordered_params=[u'projectId', u'location'],
-        path_params=[u'location', u'projectId'],
-        query_params=[u'filter', u'pageSize', u'pageToken', u'view'],
-        relative_path=u'v1b3/projects/{projectId}/locations/{location}/jobs',
-        request_field='',
-        request_type_name=u'DataflowProjectsLocationsJobsListRequest',
-        response_type_name=u'ListJobsResponse',
-        supports_download=False,
-    )
-
-    def Update(self, request, global_params=None):
-      """Updates the state of an existing Cloud Dataflow job.
-
-      Args:
-        request: (DataflowProjectsLocationsJobsUpdateRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (Job) The response message.
-      """
-      config = self.GetMethodConfig('Update')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    Update.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'PUT',
-        method_id=u'dataflow.projects.locations.jobs.update',
-        ordered_params=[u'projectId', u'location', u'jobId'],
-        path_params=[u'jobId', u'location', u'projectId'],
-        query_params=[],
-        relative_path=u'v1b3/projects/{projectId}/locations/{location}/jobs/{jobId}',
-        request_field=u'job',
-        request_type_name=u'DataflowProjectsLocationsJobsUpdateRequest',
-        response_type_name=u'Job',
-        supports_download=False,
-    )
-
-  class ProjectsLocationsService(base_api.BaseApiService):
-    """Service class for the projects_locations resource."""
-
-    _NAME = u'projects_locations'
-
-    def __init__(self, client):
-      super(DataflowV1b3.ProjectsLocationsService, self).__init__(client)
-      self._upload_configs = {
-          }
-
-  class ProjectsTemplatesService(base_api.BaseApiService):
-    """Service class for the projects_templates resource."""
-
-    _NAME = u'projects_templates'
-
-    def __init__(self, client):
-      super(DataflowV1b3.ProjectsTemplatesService, self).__init__(client)
-      self._upload_configs = {
-          }
-
-    def Create(self, request, global_params=None):
-      """Creates a Cloud Dataflow job from a template.
-
-      Args:
-        request: (DataflowProjectsTemplatesCreateRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (Job) The response message.
-      """
-      config = self.GetMethodConfig('Create')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    Create.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'POST',
-        method_id=u'dataflow.projects.templates.create',
-        ordered_params=[u'projectId'],
-        path_params=[u'projectId'],
-        query_params=[],
-        relative_path=u'v1b3/projects/{projectId}/templates',
-        request_field=u'createJobFromTemplateRequest',
-        request_type_name=u'DataflowProjectsTemplatesCreateRequest',
-        response_type_name=u'Job',
-        supports_download=False,
-    )
-
-  class ProjectsService(base_api.BaseApiService):
-    """Service class for the projects resource."""
-
-    _NAME = u'projects'
-
-    def __init__(self, client):
-      super(DataflowV1b3.ProjectsService, self).__init__(client)
-      self._upload_configs = {
-          }
-
-    def WorkerMessages(self, request, global_params=None):
-      """Send a worker_message to the service.
-
-      Args:
-        request: (DataflowProjectsWorkerMessagesRequest) input message
-        global_params: (StandardQueryParameters, default: None) global arguments
-      Returns:
-        (SendWorkerMessagesResponse) The response message.
-      """
-      config = self.GetMethodConfig('WorkerMessages')
-      return self._RunMethod(
-          config, request, global_params=global_params)
-
-    WorkerMessages.method_config = lambda: base_api.ApiMethodInfo(
-        http_method=u'POST',
-        method_id=u'dataflow.projects.workerMessages',
-        ordered_params=[u'projectId'],
-        path_params=[u'projectId'],
-        query_params=[],
-        relative_path=u'v1b3/projects/{projectId}/WorkerMessages',
-        request_field=u'sendWorkerMessagesRequest',
-        request_type_name=u'DataflowProjectsWorkerMessagesRequest',
-        response_type_name=u'SendWorkerMessagesResponse',
-        supports_download=False,
-    )


Mime
View raw message