Add base future package to google.cloud

core/google/cloud/_helpers.py

@@ -379,6 +379,29 @@ def _bytes_to_unicode(value):
         raise ValueError('%r could not be converted to unicode' % (value,))
 
 
+def _from_any_pb(pb_type, any_pb):
+    """Converts an Any protobuf to the specified message type
+
+    Args:
+        pb_type (type): the type of the message that any_pb stores an instance
+            of.
+        any_pb (google.protobuf.any_pb2.Any): the object to be converted.
+
+    Returns:
+        pb_type: An instance of the pb_type message.
+
+    Raises:
+        TypeError: if the message could not be converted.
+    """
+    msg = pb_type()
+    if not any_pb.Unpack(msg):
+        raise TypeError(
+            'Could not convert {} to {}'.format(
+                any_pb.__class__.__name__, pb_type.__name__))
+
+    return msg
+
+
 def _pb_timestamp_to_datetime(timestamp_pb):
     """Convert a Timestamp protobuf to a datetime object.
 

core/google/cloud/future/__init__.py

@@ -0,0 +1,21 @@
+# Copyright 2017, 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 dealing with asynchronous operations."""
+
+from google.cloud.future.base import Future
+
+__all__ = [
+    'Future',
+]

core/google/cloud/future/_helpers.py

@@ -0,0 +1,39 @@
+# Copyright 2017, 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.
+
+"""Private helpers for futures."""
+
+import logging
+import threading
+
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def start_daemon_thread(*args, **kwargs):
+    """Starts a thread and marks it as a daemon thread."""
+    thread = threading.Thread(*args, **kwargs)
+    thread.daemon = True
+    thread.start()
+    return thread
+
+
+def safe_invoke_callback(callback, *args, **kwargs):
+    """Invoke a callback, swallowing and logging any exceptions."""
+    # pylint: disable=bare-except
+    # We intentionally want to swallow all exceptions.
+    try:
+        return callback(*args, **kwargs)
+    except:
+        _LOGGER.exception('Error while executing Future callback.')

core/google/cloud/future/base.py

@@ -0,0 +1,173 @@
+# Copyright 2017, 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.
+
+"""Abstract and helper bases for Future implementations."""
+
+import abc
+
+import six
+
+from google.cloud.future import _helpers
+
+
+@six.add_metaclass(abc.ABCMeta)
+class Future(object):
+    # pylint: disable=missing-docstring, invalid-name
+    # We inherit the interfaces here from concurrent.futures.
+
+    """Future interface.
+
+    This interface is based on :class:`concurrent.futures.Future`.
+    """
+
+    @abc.abstractmethod
+    def cancel(self):  # pragma: NO COVER
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def cancelled(self):  # pragma: NO COVER
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def running(self):  # pragma: NO COVER
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def done(self):  # pragma: NO COVER
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def result(self, timeout=None):  # pragma: NO COVER
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def exception(self, timeout=None):  # pragma: NO COVER
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def add_done_callback(self, fn):  # pragma: NO COVER
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def set_result(self, result):  # pragma: NO COVER
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def set_exception(self, exception):  # pragma: NO COVER
+        raise NotImplementedError()
+
+
+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.
+
+    .. note: Privacy here is intended to prevent the final class from
+    overexposing, not to prevent subclasses from accessing methods.
+    """
+    def __init__(self):
+        super(PollingFuture, self).__init__()
+        self._result = None
+        self._exception = None
+        self._result_set = False
+        """bool: Set to True when the result has been set via set_result or
+        set_exception."""
+        self._polling_thread = None
+        self._done_callbacks = []
+
+    @abc.abstractmethod
+    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.
+        """
+        raise NotImplementedError()
+
+    def result(self, timeout=None):
+        """Get the result of the operation, blocking if necessary.
+
+        Args:
+            timeout (int): How long to wait for the operation to complete.
+                If None, wait indefinitely.
+
+        Returns:
+            google.protobuf.Message: The Operation's result.
+
+        Raises:
+            google.gax.GaxError: If the operation errors or if the timeout is
+                reached before the operation completes.
+        """
+        self._blocking_poll()
+
+        if self._exception is not None:
+            # pylint: disable=raising-bad-type
+            # Pylint doesn't recognize that this is valid in this case.
+            raise self._exception
+
+        return self._result
+
+    def exception(self, timeout=None):
+        """Get the exception from the operation, blocking if necessary.
+
+        Args:
+            timeout (int): How long to wait for the operation to complete.
+                If None, wait indefinitely.
+
+        Returns:
+            Optional[google.gax.GaxError]: The operation's error.
+        """
+        self._blocking_poll()
+        return self._exception
+
+    def add_done_callback(self, fn):
+        """Add a callback to be executed when the operation is complete.
+
+        If the operation is not already complete, this will start a helper
+        thread to poll for the status of the operation in the background.
+
+        Args:
+            fn (Callable[Future]): The callback to execute when the operation
+                is complete.
+        """
+        if self._result_set:
+            _helpers.safe_invoke_callback(fn, self)
+            return
+
+        self._done_callbacks.append(fn)
+
+        if self._polling_thread is None:
+            # The polling thread will exit on its own as soon as the operation
+            # is done.
+            self._polling_thread = _helpers.start_daemon_thread(
+                target=self._blocking_poll)
+
+    def _invoke_callbacks(self, *args, **kwargs):
+        """Invoke all done callbacks."""
+        for callback in self._done_callbacks:
+            _helpers.safe_invoke_callback(callback, *args, **kwargs)
+
+    def set_result(self, result):
+        """Set the Future's result."""
+        self._result = result
+        self._result_set = True
+        self._invoke_callbacks(self)
+
+    def set_exception(self, exception):
+        """Set the Future's exception."""
+        self._exception = exception
+        self._result_set = True
+        self._invoke_callbacks(self)

core/tests/unit/future/test__helpers.py

@@ -0,0 +1,52 @@
+# Copyright 2017, Google Inc.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+#     * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+#     * Redistributions in binary form must reproduce the above
+# copyright notice, this list of conditions and the following disclaimer
+# in the documentation and/or other materials provided with the
+# distribution.
+#     * Neither the name of Google Inc. nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+import mock
+
+from google.cloud.future import _helpers
+
+
+@mock.patch('threading.Thread', autospec=True)
+def test_start_deamon_thread(unused_thread):
+    deamon_thread = _helpers.start_daemon_thread(target=mock.sentinel.target)
+    assert deamon_thread.daemon is True
+
+
+def test_safe_invoke_callback():
+    callback = mock.Mock(spec=['__call__'], return_value=42)
+    result = _helpers.safe_invoke_callback(callback, 'a', b='c')
+    assert result == 42
+    callback.assert_called_once_with('a', b='c')
+
+
+def test_safe_invoke_callback_exception():
+    callback = mock.Mock(spec=['__call__'], side_effect=ValueError())
+    result = _helpers.safe_invoke_callback(callback, 'a', b='c')
+    assert result is None
+    callback.assert_called_once_with('a', b='c')