Merge branch 'master' into hv_issue719-job-manager-threaded-job-start

Author: soxofaan

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - `openeo.testing.io.TestDataLoader`: unit test utility to compactly load (and optionally preprocess) tests data (text/JSON/...)
+- `openeo.Connection`: automatically retry API requests on `429 Too Many Requests` HTTP errors, with appropriate delay if possible ([#441](https://github.com/Open-EO/openeo-python-client/issues/441))
 
 ### Changed
 

docs/basics.rst

@@ -317,7 +317,7 @@ and observations are filtered out:
 
 .. image:: _static/images/basics/evi-masked-composite.png
 
-.. _aggregate-spatial-evi
+.. _aggregate-spatial-evi:
 
 Aggregated EVI timeseries
 ===========================

docs/conf.py

@@ -41,6 +41,7 @@
     'sphinx.ext.viewcode',
     'sphinx.ext.doctest',
     'myst_parser',
+    "sphinx.ext.intersphinx",
 ]
 
 import sphinx_autodoc_typehints
@@ -194,3 +195,13 @@
      author, 'openeo', 'One line description of project.',
      'Miscellaneous'),
 ]
+
+
+# Mapping for external documentation
+intersphinx_mapping = {
+    "python": ("https://docs.python.org/3", None),
+    "numpy": ("https://numpy.org/doc/stable/", None),
+    "xarray": ("https://docs.xarray.dev/en/stable/", None),
+    "pandas": ("https://pandas.pydata.org/pandas-docs/stable/", None),
+    "urllib3": ("https://urllib3.readthedocs.io/en/stable/", None),
+}

docs/cookbook/job_manager.rst

@@ -1,4 +1,4 @@
-.. _job-manager
+.. _job-manager:
 
 ====================================
 Multi Backend Job Manager

docs/datacube_construction.rst

@@ -200,6 +200,7 @@ TODO
 
 
 .. _multi-result-process-graphs:
+
 Building process graphs with multiple result nodes
 ===================================================
 

docs/udf.rst

@@ -374,11 +374,10 @@ the datacube.
 
 
 .. warning::
-
-The ``apply_neighborhood`` is the most versatile, but also most complex process. Make sure to keep an eye on the dimensions
-and the shape of the DataArray returned by your UDF. For instance, a very common error is to somehow 'flip' the spatial dimensions.
-Debugging the UDF locally can help, but then you will want to try and reproduce the input that you get also on the backend.
-This can typically be achieved by using logging to inspect the DataArrays passed into your UDF backend side.
+    The ``apply_neighborhood`` is the most versatile, but also most complex process. Make sure to keep an eye on the dimensions
+    and the shape of the DataArray returned by your UDF. For instance, a very common error is to somehow 'flip' the spatial dimensions.
+    Debugging the UDF locally can help, but then you will want to try and reproduce the input that you get also on the backend.
+    This can typically be achieved by using logging to inspect the DataArrays passed into your UDF backend side.
 
 
 

openeo/extra/job_management/__init__.py

@@ -29,7 +29,8 @@
 import shapely.errors
 import shapely.geometry.base
 import shapely.wkt
-from requests.adapters import HTTPAdapter, Retry
+from requests.adapters import HTTPAdapter
+from urllib3.util import Retry
 
 from openeo import BatchJob, Connection
 from openeo.extra.job_management._thread_worker import (
@@ -292,7 +293,7 @@ def _make_resilient(connection):
         503 Service Unavailable
         504 Gateway Timeout
         """
-        # TODO: refactor this helper out of this class and unify with `openeo_driver.util.http.requests_with_retry`
+        # TODO: migrate this to now built-in retry configuration of `Connection` or `openeo.util.http.retry_adapter`?
         status_forcelist = [500, 502, 503, 504]
         retries = Retry(
             total=MAX_RETRIES,

openeo/rest/_connection.py

@@ -5,13 +5,15 @@
 from typing import Iterable, Optional, Union
 
 import requests
+import urllib3.util
 from requests import Response
 from requests.auth import AuthBase
 
 import openeo
 from openeo.rest import OpenEoApiError, OpenEoApiPlainError, OpenEoRestError
 from openeo.rest.auth.auth import NullAuth
 from openeo.util import ContextTimer, ensure_list, str_truncate, url_join
+from openeo.utils.http import HTTP_502_BAD_GATEWAY, session_with_retries
 
 _log = logging.getLogger(__name__)
 
@@ -26,15 +28,22 @@ class RestApiConnection:
     def __init__(
         self,
         root_url: str,
+        *,
         auth: Optional[AuthBase] = None,
         session: Optional[requests.Session] = None,
         default_timeout: Optional[int] = None,
         slow_response_threshold: Optional[float] = None,
+        retry: Union[urllib3.util.Retry, dict, bool, None] = None,
     ):
         self._root_url = root_url
         self._auth = None
         self.auth = auth or NullAuth()
-        self.session = session or requests.Session()
+        if session:
+            self.session = session
+        elif retry is not False:
+            self.session = session_with_retries(retry=retry)
+        else:
+            self.session = requests.Session()
         self.default_timeout = default_timeout or DEFAULT_TIMEOUT
         self.default_headers = {
             "User-Agent": "openeo-python-client/{cv} {py}/{pv} {pl}".format(
@@ -165,7 +174,7 @@ def _raise_api_error(self, response: requests.Response):
         _log.warning(f"Failed to parse API error response: [{status_code}] {text!r} (headers: {response.headers})")
 
         # TODO: eliminate this VITO-backend specific error massaging?
-        if status_code == 502 and "Proxy Error" in text:
+        if status_code == HTTP_502_BAD_GATEWAY and "Proxy Error" in text:
             error_message = (
                 "Received 502 Proxy Error."
                 " This typically happens when a synchronous openEO processing request takes too long and is aborted."

openeo/rest/_testing.py

@@ -17,6 +17,7 @@
 
 from openeo import Connection, DataCube
 from openeo.rest.vectorcube import VectorCube
+from openeo.utils.http import HTTP_201_CREATED, HTTP_202_ACCEPTED, HTTP_204_NO_CONTENT
 
 OPENEO_BACKEND = "https://openeo.test/"
 
@@ -212,7 +213,7 @@ def _handle_post_jobs(self, request, context):
         for field in self.extra_job_metadata_fields:
             job_data[field] = post_data.get(field)
         self.batch_jobs[job_id] = job_data
-        context.status_code = 201
+        context.status_code = HTTP_201_CREATED
         context.headers["openeo-identifier"] = job_id
 
     def _get_job_id(self, request) -> str:
@@ -259,7 +260,7 @@ def _handle_post_job_results(self, request, context):
             self.batch_jobs[job_id]["status"] = self._get_job_status(
                 job_id=job_id, current_status=self.batch_jobs[job_id]["status"]
             )
-            context.status_code = 202
+            context.status_code = HTTP_202_ACCEPTED
         else:
             self._set_job_status(job_id=job_id, status="error")
             context.status_code = failure["status_code"]
@@ -300,7 +301,7 @@ def _handle_delete_job_results(self, request, context):
         """Handler of `DELETE /job/{job_id}/results` (cancel job)."""
         job_id = self._get_job_id(request)
         self._set_job_status(job_id=job_id, status="canceled")
-        context.status_code = 204
+        context.status_code = HTTP_204_NO_CONTENT
 
     def _handle_get_job_result_asset(self, request, context):
         """Handler of `GET /job/{job_id}/results/result.data` (get batch job result asset)."""

openeo/rest/connection.py

@@ -29,6 +29,7 @@
 
 import requests
 import shapely.geometry.base
+import urllib3.util
 from requests.auth import AuthBase, HTTPBasicAuth
 
 import openeo
@@ -85,6 +86,11 @@
     load_json_resource,
     rfc3339,
 )
+from openeo.utils.http import (
+    HTTP_201_CREATED,
+    HTTP_401_UNAUTHORIZED,
+    HTTP_403_FORBIDDEN,
+)
 from openeo.utils.version import ComparableVersion
 
 __all__ = ["Connection", "connect"]
@@ -114,6 +120,16 @@ class Connection(RestApiConnection):
         optional :class:`OidcAuthenticator` object to use for renewing OIDC tokens.
     :param auth: Optional ``requests.auth.AuthBase`` object to use for requests.
         Usage of this parameter is deprecated, use the specific authentication methods instead.
+    :param retry: general request retry settings, can be specified as:
+
+        - :py:class:`urllib3.util.Retry` object
+          or a dictionary with corresponding keyword arguments
+          (e.g. ``total``, ``backoff_factor``, ``status_forcelist``, ...)
+        - ``None`` (default) to use default openEO-oriented retry settings
+        - ``False`` to disable retrying requests
+
+    .. versionchanged:: 0.41.0
+        Added ``retry`` argument.
     """
 
     _MINIMUM_API_VERSION = ComparableVersion("1.0.0")
@@ -130,6 +146,7 @@ def __init__(
         refresh_token_store: Optional[RefreshTokenStore] = None,
         oidc_auth_renewer: Optional[OidcAuthenticator] = None,
         auth: Optional[AuthBase] = None,
+        retry: Union[urllib3.util.Retry, dict, bool, None] = None,
     ):
         if "://" not in url:
             url = "https://" + url
@@ -139,6 +156,7 @@ def __init__(
             root_url=self.version_discovery(url, session=session, timeout=default_timeout),
             auth=auth, session=session, default_timeout=default_timeout,
             slow_response_threshold=slow_response_threshold,
+            retry=retry,
         )
 
         # Initial API version check.
@@ -663,7 +681,10 @@ def _request():
             # Initial request attempt
             return _request()
         except OpenEoApiError as api_exc:
-            if api_exc.http_status_code in {401, 403} and api_exc.code == "TokenInvalid":
+            if (
+                api_exc.http_status_code in {HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN}
+                and api_exc.code == "TokenInvalid"
+            ):
                 # Auth token expired: can we refresh?
                 if isinstance(self.auth, OidcBearerAuth) and self._oidc_auth_renewer:
                     msg = f"OIDC access token expired ({api_exc.http_status_code} {api_exc.code})."
@@ -1750,7 +1771,7 @@ def create_job(
         )
 
         self._preflight_validation(pg_with_metadata=pg_with_metadata, validate=validate)
-        response = self.post("/jobs", json=pg_with_metadata, expected_status=201)
+        response = self.post("/jobs", json=pg_with_metadata, expected_status=HTTP_201_CREATED)
 
         job_id = None
         if "openeo-identifier" in response.headers:
@@ -1885,6 +1906,7 @@ def connect(
     session: Optional[requests.Session] = None,
     default_timeout: Optional[int] = None,
     auto_validate: bool = True,
+    retry: Union[urllib3.util.Retry, dict, bool, None] = None,
 ) -> Connection:
     """
     This method is the entry point to OpenEO.
@@ -1904,9 +1926,19 @@ def connect(
     :param auth_options: Options/arguments specific to the authentication type
     :param default_timeout: default timeout (in seconds) for requests
     :param auto_validate: toggle to automatically validate process graphs before execution
+    :param retry: general request retry settings, can be specified as:
 
-    .. versionadded:: 0.24.0
-        added ``auto_validate`` argument
+        - :py:class:`urllib3.util.Retry` object
+          or a dictionary with corresponding keyword arguments
+          (e.g. ``total``, ``backoff_factor``, ``status_forcelist``, ...)
+        - ``None`` (default) to use default openEO-oriented retry settings
+        - ``False`` to disable retrying requests
+
+    .. versionchanged:: 0.24.0
+        Added ``auto_validate`` argument
+
+    .. versionchanged:: 0.41.0
+        Added ``retry`` argument.
     """
 
     def _config_log(message):
@@ -1931,7 +1963,13 @@ def _config_log(message):
 
     if not url:
         raise OpenEoClientException("No openEO back-end URL given or known to connect to.")
-    connection = Connection(url, session=session, default_timeout=default_timeout, auto_validate=auto_validate)
+    connection = Connection(
+        url,
+        session=session,
+        default_timeout=default_timeout,
+        auto_validate=auto_validate,
+        retry=retry,
+    )
 
     auth_type = auth_type.lower() if isinstance(auth_type, str) else auth_type
     if auth_type in {None, False, 'null', 'none'}: