Merge remote-tracking branch 'origin/master' into issue801-py314

Author: soxofaan

CHANGELOG.md

@@ -13,8 +13,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Removed
 
+- Remove unused/outdated `XarrayDataCube.plot()` and its related matplotlib dependency ([#472](https://github.com/Open-EO/openeo-python-client/issues/472))
+
 ### Fixed
 
+- `DataCube.sar_backscatter()`: add corresponding band names to metadata when enabling "mask", "contributing_area", "local_incidence_angle" or "ellipsoid_incidence_angle" ([#804](https://github.com/Open-EO/openeo-python-client/issues/804))
+- Proactively refresh access/bearer token in `MultiBackendJobManager` before launching a job start thread ([#817](https://github.com/Open-EO/openeo-python-client/issues/817))
+- `Connection.list_services()`: Fix list access error for federation extension
+
 
 ## [0.45.0] - 2025-09-17
 

openeo/_version.py

@@ -1 +1 @@
-__version__ = "0.46.0a1"
+__version__ = "0.46.0a2"

openeo/extra/job_management/__init__.py

@@ -244,6 +244,8 @@ def __init__(
         )
         self._thread = None
         self._worker_pool = None
+        # Generic cache
+        self._cache = {}
 
     def add_backend(
         self,
@@ -650,6 +652,8 @@ def _launch_job(self, start_job, df, i, backend_name, stats: Optional[dict] = No
                         # start job if not yet done by callback
                         try:
                             job_con = job.connection
+                            # Proactively refresh bearer token (because task in thread will not be able to do that)
+                            self._refresh_bearer_token(connection=job_con)
                             task = _JobStartTask(
                                 root_url=job_con.root_url,
                                 bearer_token=job_con.auth.bearer if isinstance(job_con.auth, BearerAuth) else None,
@@ -670,6 +674,21 @@ def _launch_job(self, start_job, df, i, backend_name, stats: Optional[dict] = No
                 df.loc[i, "status"] = "skipped"
                 stats["start_job skipped"] += 1
 
+    def _refresh_bearer_token(self, connection: Connection, *, max_age: float = 60) -> None:
+        """
+        Helper to proactively refresh the bearer (access) token of the connection
+        (but not too often, based on `max_age`).
+        """
+        # TODO: be smarter about timing, e.g. by inspecting expiry of current token?
+        now = time.time()
+        key = f"connection:{id(connection)}:refresh-time"
+        if self._cache.get(key, 0) + max_age < now:
+            refreshed = connection.try_access_token_refresh()
+            if refreshed:
+                self._cache[key] = now
+            else:
+                _log.warning("Failed to proactively refresh bearer token")
+
     def _process_threadworker_updates(
         self,
         worker_pool: _JobManagerWorkerThreadPool,

openeo/metadata.py

@@ -193,8 +193,10 @@ def filter_bands(self, bands: List[Union[int, str]]) -> BandDimension:
             bands=[self.bands[self.band_index(b)] for b in bands]
         )
 
-    def append_band(self, band: Band) -> BandDimension:
+    def append_band(self, band: Union[Band, str]) -> BandDimension:
         """Create new BandDimension with appended band."""
+        if isinstance(band, str):
+            band = Band(name=band)
         if band.name in self.band_names:
             raise ValueError("Duplicate band {b!r}".format(b=band))
 
@@ -260,7 +262,10 @@ class CubeMetadata:
 
     def __init__(self, dimensions: Optional[List[Dimension]] = None):
         # Original collection metadata (actual cube metadata might be altered through processes)
-        self._dimensions = dimensions
+        # TODO: for `self._dimensions` we use `None` here to indicate an unknown/unspecified dimension set,
+        #       but most usage actually assumes it is a list that can be iterated over.
+        #       Can we handle this more consistently and less error-prone?
+        self._dimensions: Union[List[Dimension], None] = dimensions
         self._band_dimension = None
         self._temporal_dimension = None
 
@@ -283,8 +288,12 @@ def __eq__(self, o: Any) -> bool:
         return isinstance(o, type(self)) and self._dimensions == o._dimensions
 
     def __repr__(self) -> str:
-        bands = self.band_names if self.has_band_dimension() else "no bands dimension"
-        return f"{self.__class__.__name__}({bands} - {self.dimension_names()})"
+        if self.has_band_dimension():
+            return f"{self.__class__.__name__}(dimension_names={self.dimension_names()}, band_names={self.band_names})"
+        elif self._dimensions is not None:
+            return f"{self.__class__.__name__}(dimension_names={self.dimension_names()})"
+        else:
+            return f"{self.__class__.__name__}(dimensions=None)"
 
     def __str__(self) -> str:
         bands = self.band_names if self.has_band_dimension() else "no bands dimension"
@@ -297,7 +306,10 @@ def _clone_and_update(self, dimensions: Optional[List[Dimension]] = None, **kwar
             dimensions = self._dimensions
         return cls(dimensions=dimensions, **kwargs)
 
-    def dimension_names(self) -> List[str]:
+    def dimension_names(self) -> Union[List[str], None]:
+        if self._dimensions is None:
+            # TODO: better solution for unknown dimensions?
+            return None
         return list(d.name for d in self._dimensions)
 
     def assert_valid_dimension(self, dimension: str) -> str:
@@ -369,7 +381,7 @@ def filter_bands(self, band_names: List[Union[int, str]]) -> CubeMetadata:
             dimensions=[d.filter_bands(band_names) if isinstance(d, BandDimension) else d for d in self._dimensions]
         )
 
-    def append_band(self, band: Band) -> CubeMetadata:
+    def append_band(self, band: Union[Band, str]) -> CubeMetadata:
         """
         Create new `CubeMetadata` with given band added to band dimension.
         """

openeo/rest/_testing.py

@@ -132,6 +132,22 @@ def at_url(cls, root_url: str, *, requests_mock, capabilities: Optional[dict] =
         connection = Connection(root_url)
         return cls(requests_mock=requests_mock, connection=connection)
 
+    def setup_credentials_oidc(self, *, issuer: str = "https://oidc.test", id: str = "oi"):
+        self._requests_mock.get(
+            self.connection.build_url("/credentials/oidc"),
+            json={
+                "providers": [
+                    {
+                        "id": id,
+                        "issuer": issuer,
+                        "title": id,
+                        "scopes": ["openid"],
+                    }
+                ]
+            },
+        )
+        return self
+
     def setup_collection(
         self,
         collection_id: str,

openeo/rest/auth/testing.py

@@ -143,6 +143,11 @@ def token_callback_resource_owner_password_credentials(self, params: dict, conte
         assert params["scope"] == self.expected_fields["scope"]
         return self._build_token_response()
 
+    def token_callback_block_400(self, params: dict, context):
+        """Failing callback with 400 Bad Request"""
+        context.status_code = 400
+        return "block_400"
+
     def device_code_callback(self, request: requests_mock.request._RequestObjectProxy, context):
         params = self._get_query_params(query=request.text)
         assert params["client_id"] == self.expected_client_id

openeo/rest/connection.py

@@ -342,28 +342,32 @@ def _authenticate_oidc(
         *,
         provider_id: str,
         store_refresh_token: bool = False,
-        fallback_refresh_token_to_store: Optional[str] = None,
+        auto_renew_from_refresh_token: bool = False,
+        fallback_refresh_token: Optional[str] = None,
         oidc_auth_renewer: Optional[OidcAuthenticator] = None,
     ) -> Connection:
         """
         Authenticate through OIDC and set up bearer token (based on OIDC access_token) for further requests.
         """
-        tokens = authenticator.get_tokens(request_refresh_token=store_refresh_token)
+        request_refresh_token = store_refresh_token or (not oidc_auth_renewer and auto_renew_from_refresh_token)
+        tokens = authenticator.get_tokens(request_refresh_token=request_refresh_token)
         _log.info("Obtained tokens: {t}".format(t=[k for k, v in tokens._asdict().items() if v]))
+
+        refresh_token = tokens.refresh_token or fallback_refresh_token
         if store_refresh_token:
-            refresh_token = tokens.refresh_token or fallback_refresh_token_to_store
             if refresh_token:
                 self._get_refresh_token_store().set_refresh_token(
                     issuer=authenticator.provider_info.issuer,
                     client_id=authenticator.client_id,
                     refresh_token=refresh_token
                 )
-                if not oidc_auth_renewer:
-                    oidc_auth_renewer = OidcRefreshTokenAuthenticator(
-                        client_info=authenticator.client_info, refresh_token=refresh_token
-                    )
             else:
                 _log.warning("No OIDC refresh token to store.")
+        if not oidc_auth_renewer and auto_renew_from_refresh_token and refresh_token:
+            oidc_auth_renewer = OidcRefreshTokenAuthenticator(
+                client_info=authenticator.client_info, refresh_token=refresh_token
+            )
+
         token = tokens.access_token
         self.auth = OidcBearerAuth(provider_id=provider_id, access_token=token)
         self._oidc_auth_renewer = oidc_auth_renewer
@@ -452,7 +456,12 @@ def authenticate_oidc_resource_owner_password_credentials(
         authenticator = OidcResourceOwnerPasswordAuthenticator(
             client_info=client_info, username=username, password=password
         )
-        return self._authenticate_oidc(authenticator, provider_id=provider_id, store_refresh_token=store_refresh_token)
+        return self._authenticate_oidc(
+            authenticator,
+            provider_id=provider_id,
+            store_refresh_token=store_refresh_token,
+            oidc_auth_renewer=authenticator,
+        )
 
     def authenticate_oidc_refresh_token(
         self,
@@ -493,7 +502,7 @@ def authenticate_oidc_refresh_token(
             authenticator,
             provider_id=provider_id,
             store_refresh_token=store_refresh_token,
-            fallback_refresh_token_to_store=refresh_token,
+            fallback_refresh_token=refresh_token,
             oidc_auth_renewer=authenticator,
         )
 
@@ -534,7 +543,13 @@ def authenticate_oidc_device(
         authenticator = OidcDeviceAuthenticator(
             client_info=client_info, use_pkce=use_pkce, max_poll_time=max_poll_time, **kwargs
         )
-        return self._authenticate_oidc(authenticator, provider_id=provider_id, store_refresh_token=store_refresh_token)
+        return self._authenticate_oidc(
+            authenticator,
+            provider_id=provider_id,
+            store_refresh_token=store_refresh_token,
+            # TODO: expose `auto_renew_from_refresh_token` directly as option instead of reusing `store_refresh_token` arg?
+            auto_renew_from_refresh_token=store_refresh_token,
+        )
 
     def authenticate_oidc(
         self,
@@ -604,7 +619,8 @@ def authenticate_oidc(
                     authenticator,
                     provider_id=provider_id,
                     store_refresh_token=store_refresh_token,
-                    fallback_refresh_token_to_store=refresh_token,
+                    fallback_refresh_token=refresh_token,
+                    oidc_auth_renewer=authenticator,
                 )
                 # TODO: pluggable/jupyter-aware display function?
                 print("Authenticated using refresh token.")
@@ -622,6 +638,8 @@ def authenticate_oidc(
             authenticator,
             provider_id=provider_id,
             store_refresh_token=store_refresh_token,
+            # TODO: expose `auto_renew_from_refresh_token` directly as option instead of reusing `store_refresh_token` arg?
+            auto_renew_from_refresh_token=store_refresh_token,
         )
         print("Authenticated using device code flow.")
         return con
@@ -665,6 +683,28 @@ def authenticate_bearer_token(self, bearer_token: str) -> Connection:
         self._oidc_auth_renewer = None
         return self
 
+    def try_access_token_refresh(self, *, reason: Optional[str] = None) -> bool:
+        """
+        Try to get a fresh access token if possible.
+        Returns whether a new access token was obtained.
+        """
+        reason = f" Reason: {reason}" if reason else ""
+        if isinstance(self.auth, OidcBearerAuth) and self._oidc_auth_renewer:
+            try:
+                self._authenticate_oidc(
+                    authenticator=self._oidc_auth_renewer,
+                    provider_id=self._oidc_auth_renewer.provider_info.id,
+                    store_refresh_token=False,
+                    oidc_auth_renewer=self._oidc_auth_renewer,
+                )
+                _log.info(f"Obtained new access token (grant {self._oidc_auth_renewer.grant_type!r}).{reason}")
+                return True
+            except OpenEoClientException as auth_exc:
+                _log.error(
+                    f"Failed to obtain new access token (grant {self._oidc_auth_renewer.grant_type!r}): {auth_exc!r}.{reason}"
+                )
+        return False
+
     def request(
         self,
         method: str,
@@ -690,24 +730,11 @@ def _request():
                 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})."
-                    try:
-                        self._authenticate_oidc(
-                            authenticator=self._oidc_auth_renewer,
-                            provider_id=self._oidc_auth_renewer.provider_info.id,
-                            store_refresh_token=False,
-                            oidc_auth_renewer=self._oidc_auth_renewer,
-                        )
-                        _log.info(f"{msg} Obtained new access token (grant {self._oidc_auth_renewer.grant_type!r}).")
-                    except OpenEoClientException as auth_exc:
-                        _log.error(
-                            f"{msg} Failed to obtain new access token (grant {self._oidc_auth_renewer.grant_type!r}): {auth_exc!r}."
-                        )
-                    else:
-                        # Retry request.
-                        return _request()
+                # Retry if we can refresh the access token
+                if self.try_access_token_refresh(
+                    reason=f"OIDC access token expired ({api_exc.http_status_code} {api_exc.code})."
+                ):
+                    return _request()
             raise
 
     def describe_account(self) -> dict:
@@ -821,12 +848,12 @@ def list_services(self) -> list:
         :return: data_dict: Dict All available services
         """
         # TODO return parsed service objects
-        services = self.get('/services', expected_status=200).json()["services"]
-        federation_missing = federation_extension.get_federation_missing(data=services, resource_name="services")
+        response = self.get("/services", expected_status=200).json()
+        federation_missing = federation_extension.get_federation_missing(data=response, resource_name="services")
         federation = self.capabilities().ext_federation_backend_details()
         return VisualList(
             "data-table",
-            data=services,
+            data=response["services"],
             parameters={"columns": "services", "missing": federation_missing, "federation": federation},
         )
 

openeo/rest/datacube.py

@@ -2943,7 +2943,16 @@ def sar_backscatter(
         }
         if options:
             arguments["options"] = options
-        return self.process(process_id="sar_backscatter", arguments=arguments)
+        metadata = self.metadata
+        if mask:
+            metadata = metadata.append_band(band="mask")
+        if contributing_area:
+            metadata = metadata.append_band(band="contributing_area")
+        if local_incidence_angle:
+            metadata = metadata.append_band(band="local_incidence_angle")
+        if ellipsoid_incidence_angle:
+            metadata = metadata.append_band(band="ellipsoid_incidence_angle")
+        return self.process(process_id="sar_backscatter", arguments=arguments, metadata=metadata)
 
     @openeo_process
     def fit_curve(self, parameters: list, function: Union[str, PGNode, typing.Callable], dimension: str):