Add operation future

core/google/cloud/future/base.py

@@ -15,8 +15,12 @@
 """Abstract and helper bases for Future implementations."""
 
 import abc
+import concurrent.futures
+import functools
+import operator
 
 import six
+import tenacity
 
 from google.cloud.future import _helpers
 
@@ -72,8 +76,8 @@ def set_exception(self, exception):
 class PollingFuture(Future):
     """A Future that needs to poll some service to check its status.
 
-    The private :meth:`_blocking_poll` method should be implemented by
-    subclasses.
+    The :meth:`done` method should be implemented by subclasses. The polling
+    behavior will repeatedly call ``done`` until it returns True.
 
     .. note: Privacy here is intended to prevent the final class from
     overexposing, not to prevent subclasses from accessing methods.
@@ -89,15 +93,52 @@ def __init__(self):
         self._done_callbacks = []
 
     @abc.abstractmethod
+    def done(self):
+        """Checks to see if the operation is complete.
+
+        Returns:
+            bool: True if the operation is complete, False otherwise.
+        """
+        # pylint: disable=redundant-returns-doc, missing-raises-doc
+        raise NotImplementedError()
+
+    def running(self):
+        """True if the operation is currently running."""
+        return not self.done()
+
     def _blocking_poll(self, timeout=None):
         """Poll and wait for the Future to be resolved.
 
         Args:
             timeout (int): How long to wait for the operation to complete.
                 If None, wait indefinitely.
         """
-        # pylint: disable=missing-raises
-        raise NotImplementedError()
+        if self._result_set:
+            return
+
+        retry_on = tenacity.retry_if_result(
+            functools.partial(operator.is_not, True))
+        # Use exponential backoff with jitter.
+        wait_on = (
+            tenacity.wait_exponential(multiplier=1, max=10) +
+            tenacity.wait_random(0, 1))
+
+        if timeout is None:
+            retry = tenacity.retry(retry=retry_on, wait=wait_on)
+        else:
+            retry = tenacity.retry(
+                retry=retry_on,
+                wait=wait_on,
+                stop=tenacity.stop_after_delay(timeout))
+
+        try:
+            retry(self.done)()
+        except tenacity.RetryError as exc:
+            six.raise_from(
+                concurrent.futures.TimeoutError(
+                    'Operation did not complete within the designated '
+                    'timeout.'),
+                exc)
 
     def result(self, timeout=None):
         """Get the result of the operation, blocking if necessary.
@@ -113,7 +154,7 @@ def result(self, timeout=None):
             google.gax.GaxError: If the operation errors or if the timeout is
                 reached before the operation completes.
         """
-        self._blocking_poll()
+        self._blocking_poll(timeout=timeout)
 
         if self._exception is not None:
             # pylint: disable=raising-bad-type

core/google/cloud/future/operation.py

@@ -0,0 +1,247 @@
+# Copyright 2016 Google Inc.
+#
+# 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.
+
+"""Futures for long-running operations returned from Google Cloud APIs."""
+
+import functools
+import threading
+
+from google.longrunning import operations_pb2
+from google.protobuf import json_format
+from google.rpc import code_pb2
+
+from google.cloud import _helpers
+from google.cloud import exceptions
+from google.cloud.future import base
+
+
+class Operation(base.PollingFuture):
+    """A Future for interacting with a Google API Long-Running Operation.
+
+    Args:
+        operation (google.longrunning.operations_pb2.Operation): The
+            initial operation.
+        refresh (Callable[[], Operation]): A callable that returns the
+            latest state of the operation.
+        cancel (Callable[[], None]), A callable that tries to cancel
+            the operation.
+        result_type (type): The protobuf type for the operation's result.
+        metadata_type (type): The protobuf type for the operation's
+            metadata.
+    """
+
+    def __init__(
+            self, operation, refresh, cancel,
+            result_type, metadata_type=None):
+        super(Operation, self).__init__()
+        self._operation = operation
+        self._refresh = refresh
+        self._cancel = cancel
+        self._result_type = result_type
+        self._metadata_type = metadata_type
+        self._completion_lock = threading.Lock()
+        # Invoke this in case the operation came back already complete.
+        self._set_result_from_operation()
+
+    @property
+    def operation(self):
+        """google.longrunning.Operation: The current long-running operation."""
+        return self._operation
+
+    @property
+    def metadata(self):
+        """google.protobuf.Message: the current operation metadata."""
+        if not self._operation.HasField('metadata'):
+            return None
+
+        return _helpers._from_any_pb(
+            self._metadata_type, self._operation.metadata)
+
+    def _set_result_from_operation(self):
+        """Set the result or exception from the operation if it is complete."""
+        # This must be done in a lock to prevent the polling thread
+        # and main thread from both executing the completion logic
+        # at the same time.
+        with self._completion_lock:
+            # If the operation isn't complete or if the result has already been
+            # set, do not call set_result/set_exception again.
+            # Note: self._result_set is set to True in set_result and
+            # set_exception, in case those methods are invoked directly.
+            if not self._operation.done or self._result_set:
+                return
+
+            if self._operation.HasField('response'):
+                response = _helpers._from_any_pb(
+                    self._result_type, self._operation.response)
+                self.set_result(response)
+            elif self._operation.HasField('error'):
+                exception = exceptions.GoogleCloudError(
+                    self._operation.error.message,
+                    errors=(self._operation.error))
+                self.set_exception(exception)
+            else:
+                exception = exceptions.GoogleCloudError(
+                    'Unexpected state: Long-running operation had neither '
+                    'response nor error set.')
+                self.set_exception(exception)
+
+    def _refresh_and_update(self):
+        """Refresh the operation and update the result if needed."""
+        # If the currently cached operation is done, no need to make another
+        # RPC as it will not change once done.
+        if not self._operation.done:
+            self._operation = self._refresh()
+            self._set_result_from_operation()
+
+    def done(self):
+        """Checks to see if the operation is complete.
+
+        Returns:
+            bool: True if the operation is complete, False otherwise.
+        """
+        self._refresh_and_update()
+        return self._operation.done
+
+    def cancel(self):
+        """Attempt to cancel the operation.
+
+        Returns:
+            bool: True if the cancel RPC was made, False if the operation is
+                already complete.
+        """
+        if self.done():
+            return False
+
+        self._cancel()
+        return True
+
+    def cancelled(self):
+        """True if the operation was cancelled."""
+        self._refresh_and_update()
+        return (self._operation.HasField('error') and
+                self._operation.error.code == code_pb2.CANCELLED)
+
+
+def _refresh_http(api_request, operation_name):
+    """Refresh an operation using a JSON/HTTP client.
+
+    Args:
+        api_request (Callable): A callable used to make an API request. This
+            should generally be
+            :meth:`google.cloud._http.Connection.api_request`.
+        operation_name (str): The name of the operation.
+
+    Returns:
+        google.longrunning.operations_pb2.Operation: The operation.
+    """
+    path = 'operations/{}'.format(operation_name)
+    api_response = api_request(method='GET', path=path)
+    return json_format.ParseDict(
+        api_response, operations_pb2.Operation())
+
+
+def _cancel_http(api_request, operation_name):
+    """Cancel an operation using a JSON/HTTP client.
+
+    Args:
+        api_request (Callable): A callable used to make an API request. This
+            should generally be
+            :meth:`google.cloud._http.Connection.api_request`.
+        operation_name (str): The name of the operation.
+    """
+    path = 'operations/{}:cancel'.format(operation_name)
+    api_request(method='POST', path=path)
+
+
+def from_http_json(operation, api_request, result_type, **kwargs):
+    """Create an operation future from using a HTTP/JSON client.
+
+    This interacts with the long-running operations `service`_ (specific
+    to a given API) vis `HTTP/JSON`_.
+
+    .. _HTTP/JSON: https://cloud.google.com/speech/reference/rest/\
+            v1beta1/operations#Operation
+
+    Args:
+        operation (dict): Operation as a dictionary.
+        api_request (Callable): A callable used to make an API request. This
+            should generally be
+            :meth:`google.cloud._http.Connection.api_request`.
+        result_type (type): The protobuf result type.
+        kwargs: Keyword args passed into the :class:`Operation` constructor.
+
+    Returns:
+        Operation: The operation future to track the given operation.
+    """
+    operation_proto = json_format.ParseDict(
+        operation, operations_pb2.Operation())
+    refresh = functools.partial(
+        _refresh_http, api_request, operation_proto.name)
+    cancel = functools.partial(
+        _cancel_http, api_request, operation_proto.name)
+    return Operation(operation_proto, refresh, cancel, result_type, **kwargs)
+
+
+def _refresh_grpc(operations_stub, operation_name):
+    """Refresh an operation using a gRPC client.
+
+    Args:
+        operations_stub (google.longrunning.operations_pb2.OperationsStub):
+            The gRPC operations stub.
+        operation_name (str): The name of the operation.
+
+    Returns:
+        google.longrunning.operations_pb2.Operation: The operation.
+    """
+    request_pb = operations_pb2.GetOperationRequest(name=operation_name)
+    return operations_stub.GetOperation(request_pb)
+
+
+def _cancel_grpc(operations_stub, operation_name):
+    """Cancel an operation using a gRPC client.
+
+    Args:
+        operations_stub (google.longrunning.operations_pb2.OperationsStub):
+            The gRPC operations stub.
+        operation_name (str): The name of the operation.
+    """
+    request_pb = operations_pb2.CancelOperationRequest(name=operation_name)
+    operations_stub.CancelOperation(request_pb)
+
+
+def from_grpc(operation, operations_stub, result_type, **kwargs):
+    """Create an operation future from using a gRPC client.
+
+    This interacts with the long-running operations `service`_ (specific
+    to a given API) via gRPC.
+
+    .. _service: https://github.com/googleapis/googleapis/blob/\
+                 050400df0fdb16f63b63e9dee53819044bffc857/\
+                 google/longrunning/operations.proto#L38
+
+    Args:
+        operation (google.longrunning.operations_pb2.Operation): The operation.
+        operations_stub (google.longrunning.operations_pb2.OperationsStub):
+            The operations stub.
+        result_type (type): The protobuf result type.
+        kwargs: Keyword args passed into the :class:`Operation` constructor.
+
+    Returns:
+        Operation: The operation future to track the given operation.
+    """
+    refresh = functools.partial(
+        _refresh_grpc, operations_stub, operation.name)
+    cancel = functools.partial(
+        _cancel_grpc, operations_stub, operation.name)
+    return Operation(operation, refresh, cancel, result_type, **kwargs)

core/setup.py

@@ -57,6 +57,7 @@
     'google-auth >= 0.4.0, < 2.0.0dev',
     'google-auth-httplib2',
     'six',
+    'tenacity >= 4.0.0, <5.0.0dev'
 ]
 
 setup(