Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Copyright 2017, Google LLC
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
15"""Abstract and helper bases for Future implementations."""
17import abc
18import concurrent.futures
20from google.api_core import exceptions
21from google.api_core import retry as retries
22from google.api_core.future import _helpers, base
25class _OperationNotComplete(Exception):
26 """Private exception used for polling via retry."""
28 pass
31# DEPRECATED as it conflates RPC retry and polling concepts into one.
32# Use POLLING_PREDICATE instead to configure polling.
33RETRY_PREDICATE = retries.if_exception_type(
34 _OperationNotComplete,
35 exceptions.TooManyRequests,
36 exceptions.InternalServerError,
37 exceptions.BadGateway,
38 exceptions.ServiceUnavailable,
39)
41# DEPRECATED: use DEFAULT_POLLING to configure LRO polling logic. Construct
42# Retry object using its default values as a baseline for any custom retry logic
43# (not to be confused with polling logic).
44DEFAULT_RETRY = retries.Retry(predicate=RETRY_PREDICATE)
46# POLLING_PREDICATE is supposed to poll only on _OperationNotComplete.
47# Any RPC-specific errors (like ServiceUnavailable) will be handled
48# by retry logic (not to be confused with polling logic) which is triggered for
49# every polling RPC independently of polling logic but within its context.
50POLLING_PREDICATE = retries.if_exception_type(
51 _OperationNotComplete,
52)
54# Default polling configuration
55DEFAULT_POLLING = retries.Retry(
56 predicate=POLLING_PREDICATE,
57 initial=1.0, # seconds
58 maximum=20.0, # seconds
59 multiplier=1.5,
60 timeout=900, # seconds
61)
64class PollingFuture(base.Future):
65 """A Future that needs to poll some service to check its status.
67 The :meth:`done` method should be implemented by subclasses. The polling
68 behavior will repeatedly call ``done`` until it returns True.
70 The actual polling logic is encapsulated in :meth:`result` method. See
71 documentation for that method for details on how polling works.
73 .. note::
75 Privacy here is intended to prevent the final class from
76 overexposing, not to prevent subclasses from accessing methods.
78 Args:
79 polling (google.api_core.retry.Retry): The configuration used for polling.
80 This parameter controls how often :meth:`done` is polled. If the
81 ``timeout`` argument is specified in :meth:`result` method it will
82 override the ``polling.timeout`` property.
83 retry (google.api_core.retry.Retry): DEPRECATED use ``polling`` instead.
84 If set, it will override ``polling`` parameter for backward
85 compatibility.
86 """
88 _DEFAULT_VALUE = object()
90 def __init__(self, polling=DEFAULT_POLLING, **kwargs):
91 super(PollingFuture, self).__init__()
92 self._polling = kwargs.get("retry", polling)
93 self._result = None
94 self._exception = None
95 self._result_set = False
96 """bool: Set to True when the result has been set via set_result or
97 set_exception."""
98 self._polling_thread = None
99 self._done_callbacks = []
101 @abc.abstractmethod
102 def done(self, retry=None):
103 """Checks to see if the operation is complete.
105 Args:
106 retry (google.api_core.retry.Retry): (Optional) How to retry the
107 polling RPC (to not be confused with polling configuration. See
108 the documentation for :meth:`result` for details).
110 Returns:
111 bool: True if the operation is complete, False otherwise.
112 """
113 # pylint: disable=redundant-returns-doc, missing-raises-doc
114 raise NotImplementedError()
116 def _done_or_raise(self, retry=None):
117 """Check if the future is done and raise if it's not."""
118 if not self.done(retry=retry):
119 raise _OperationNotComplete()
121 def running(self):
122 """True if the operation is currently running."""
123 return not self.done()
125 def _blocking_poll(self, timeout=_DEFAULT_VALUE, retry=None, polling=None):
126 """Poll and wait for the Future to be resolved."""
128 if self._result_set:
129 return
131 polling = polling or self._polling
132 if timeout is not PollingFuture._DEFAULT_VALUE:
133 polling = polling.with_timeout(timeout)
135 try:
136 polling(self._done_or_raise)(retry=retry)
137 except exceptions.RetryError:
138 raise concurrent.futures.TimeoutError(
139 f"Operation did not complete within the designated timeout of "
140 f"{polling.timeout} seconds."
141 )
143 def result(self, timeout=_DEFAULT_VALUE, retry=None, polling=None):
144 """Get the result of the operation.
146 This method will poll for operation status periodically, blocking if
147 necessary. If you just want to make sure that this method does not block
148 for more than X seconds and you do not care about the nitty-gritty of
149 how this method operates, just call it with ``result(timeout=X)``. The
150 other parameters are for advanced use only.
152 Every call to this method is controlled by the following three
153 parameters, each of which has a specific, distinct role, even though all three
154 may look very similar: ``timeout``, ``retry`` and ``polling``. In most
155 cases users do not need to specify any custom values for any of these
156 parameters and may simply rely on default ones instead.
158 If you choose to specify custom parameters, please make sure you've
159 read the documentation below carefully.
161 First, please check :class:`google.api_core.retry.Retry`
162 class documentation for the proper definition of timeout and deadline
163 terms and for the definition the three different types of timeouts.
164 This class operates in terms of Retry Timeout and Polling Timeout. It
165 does not let customizing RPC timeout and the user is expected to rely on
166 default behavior for it.
168 The roles of each argument of this method are as follows:
170 ``timeout`` (int): (Optional) The Polling Timeout as defined in
171 :class:`google.api_core.retry.Retry`. If the operation does not complete
172 within this timeout an exception will be thrown. This parameter affects
173 neither Retry Timeout nor RPC Timeout.
175 ``retry`` (google.api_core.retry.Retry): (Optional) How to retry the
176 polling RPC. The ``retry.timeout`` property of this parameter is the
177 Retry Timeout as defined in :class:`google.api_core.retry.Retry`.
178 This parameter defines ONLY how the polling RPC call is retried
179 (i.e. what to do if the RPC we used for polling returned an error). It
180 does NOT define how the polling is done (i.e. how frequently and for
181 how long to call the polling RPC); use the ``polling`` parameter for that.
182 If a polling RPC throws and error and retrying it fails, the whole
183 future fails with the corresponding exception. If you want to tune which
184 server response error codes are not fatal for operation polling, use this
185 parameter to control that (``retry.predicate`` in particular).
187 ``polling`` (google.api_core.retry.Retry): (Optional) How often and
188 for how long to call the polling RPC periodically (i.e. what to do if
189 a polling rpc returned successfully but its returned result indicates
190 that the long running operation is not completed yet, so we need to
191 check it again at some point in future). This parameter does NOT define
192 how to retry each individual polling RPC in case of an error; use the
193 ``retry`` parameter for that. The ``polling.timeout`` of this parameter
194 is Polling Timeout as defined in as defined in
195 :class:`google.api_core.retry.Retry`.
197 For each of the arguments, there are also default values in place, which
198 will be used if a user does not specify their own. The default values
199 for the three parameters are not to be confused with the default values
200 for the corresponding arguments in this method (those serve as "not set"
201 markers for the resolution logic).
203 If ``timeout`` is provided (i.e.``timeout is not _DEFAULT VALUE``; note
204 the ``None`` value means "infinite timeout"), it will be used to control
205 the actual Polling Timeout. Otherwise, the ``polling.timeout`` value
206 will be used instead (see below for how the ``polling`` config itself
207 gets resolved). In other words, this parameter effectively overrides
208 the ``polling.timeout`` value if specified. This is so to preserve
209 backward compatibility.
211 If ``retry`` is provided (i.e. ``retry is not None``) it will be used to
212 control retry behavior for the polling RPC and the ``retry.timeout``
213 will determine the Retry Timeout. If not provided, the
214 polling RPC will be called with whichever default retry config was
215 specified for the polling RPC at the moment of the construction of the
216 polling RPC's client. For example, if the polling RPC is
217 ``operations_client.get_operation()``, the ``retry`` parameter will be
218 controlling its retry behavior (not polling behavior) and, if not
219 specified, that specific method (``operations_client.get_operation()``)
220 will be retried according to the default retry config provided during
221 creation of ``operations_client`` client instead. This argument exists
222 mainly for backward compatibility; users are very unlikely to ever need
223 to set this parameter explicitly.
225 If ``polling`` is provided (i.e. ``polling is not None``), it will be used
226 to control the overall polling behavior and ``polling.timeout`` will
227 control Polling Timeout unless it is overridden by ``timeout`` parameter
228 as described above. If not provided, the``polling`` parameter specified
229 during construction of this future (the ``polling`` argument in the
230 constructor) will be used instead. Note: since the ``timeout`` argument may
231 override ``polling.timeout`` value, this parameter should be viewed as
232 coupled with the ``timeout`` parameter as described above.
234 Args:
235 timeout (int): (Optional) How long (in seconds) to wait for the
236 operation to complete. If None, wait indefinitely.
237 retry (google.api_core.retry.Retry): (Optional) How to retry the
238 polling RPC. This defines ONLY how the polling RPC call is
239 retried (i.e. what to do if the RPC we used for polling returned
240 an error). It does NOT define how the polling is done (i.e. how
241 frequently and for how long to call the polling RPC).
242 polling (google.api_core.retry.Retry): (Optional) How often and
243 for how long to call polling RPC periodically. This parameter
244 does NOT define how to retry each individual polling RPC call
245 (use the ``retry`` parameter for that).
247 Returns:
248 google.protobuf.Message: The Operation's result.
250 Raises:
251 google.api_core.GoogleAPICallError: If the operation errors or if
252 the timeout is reached before the operation completes.
253 """
255 self._blocking_poll(timeout=timeout, retry=retry, polling=polling)
257 if self._exception is not None:
258 # pylint: disable=raising-bad-type
259 # Pylint doesn't recognize that this is valid in this case.
260 raise self._exception
262 return self._result
264 def exception(self, timeout=_DEFAULT_VALUE):
265 """Get the exception from the operation, blocking if necessary.
267 See the documentation for the :meth:`result` method for details on how
268 this method operates, as both ``result`` and this method rely on the
269 exact same polling logic. The only difference is that this method does
270 not accept ``retry`` and ``polling`` arguments but relies on the default ones
271 instead.
273 Args:
274 timeout (int): How long to wait for the operation to complete.
275 If None, wait indefinitely.
277 Returns:
278 Optional[google.api_core.GoogleAPICallError]: The operation's
279 error.
280 """
281 self._blocking_poll(timeout=timeout)
282 return self._exception
284 def add_done_callback(self, fn):
285 """Add a callback to be executed when the operation is complete.
287 If the operation is not already complete, this will start a helper
288 thread to poll for the status of the operation in the background.
290 Args:
291 fn (Callable[Future]): The callback to execute when the operation
292 is complete.
293 """
294 if self._result_set:
295 _helpers.safe_invoke_callback(fn, self)
296 return
298 self._done_callbacks.append(fn)
300 if self._polling_thread is None:
301 # The polling thread will exit on its own as soon as the operation
302 # is done.
303 self._polling_thread = _helpers.start_daemon_thread(
304 target=self._blocking_poll
305 )
307 def _invoke_callbacks(self, *args, **kwargs):
308 """Invoke all done callbacks."""
309 for callback in self._done_callbacks:
310 _helpers.safe_invoke_callback(callback, *args, **kwargs)
312 def set_result(self, result):
313 """Set the Future's result."""
314 self._result = result
315 self._result_set = True
316 self._invoke_callbacks(self)
318 def set_exception(self, exception):
319 """Set the Future's exception."""
320 self._exception = exception
321 self._result_set = True
322 self._invoke_callbacks(self)