Issue #404/#481 introduce connection level auto_validate option

- connection level option will be more practical and user friendly
- per `execute` overriding is still possible
- generalize `_warn_if_process_graph_invalid` to `_preflight_validation` (to future proof it)
- additional finetuning

Author: soxofaan

CHANGELOG.md

@@ -9,9 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
-- Automatically validate process graph before download/execute/execute_batch and log warning when invalid ([#404](https://github.com/Open-EO/openeo-python-client/issues/404))
 
 - Add `DataCube.reduce_spatial()`
+- Added option (enabled by default) to automatically validate a process graph before execution. Validation issues just trigger warnings for now. ([#404](https://github.com/Open-EO/openeo-python-client/issues/404))
 
 ### Changed
 

openeo/rest/connection.py

@@ -260,6 +260,7 @@ def __init__(
         refresh_token_store: Optional[RefreshTokenStore] = None,
         slow_response_threshold: Optional[float] = None,
         oidc_auth_renewer: Optional[OidcAuthenticator] = None,
+        auto_validate: bool = True,
     ):
         """
         Constructor of Connection, authenticates user.
@@ -282,6 +283,7 @@ def __init__(
         self._auth_config = auth_config
         self._refresh_token_store = refresh_token_store
         self._oidc_auth_renewer = oidc_auth_renewer
+        self._auto_validate = auto_validate
 
     @classmethod
     def version_discovery(
@@ -1052,8 +1054,8 @@ def validate_process_graph(self, process_graph: dict) -> List[dict]:
         :param process_graph: (flat) dict representing process graph
         :return: list of errors (dictionaries with "code" and "message" fields)
         """
-        graph = self._build_request_with_process_graph(process_graph)["process"]
-        return self.post(path="/validation", json=graph, expected_status=200).json()["errors"]
+        pg_with_metadata = self._build_request_with_process_graph(process_graph)["process"]
+        return self.post(path="/validation", json=pg_with_metadata, expected_status=200).json()["errors"]
 
     @property
     def _api_version(self) -> ComparableVersion:
@@ -1393,8 +1395,9 @@ def load_url(self, url: str, format: str, options: Optional[dict] = None):
 
     def create_service(self, graph: dict, type: str, **kwargs) -> Service:
         # TODO: type hint for graph: is it a nested or a flat one?
-        req = self._build_request_with_process_graph(process_graph=graph, type=type, **kwargs)
-        response = self.post(path="/services", json=req, expected_status=201)
+        pg_with_metadata = self._build_request_with_process_graph(process_graph=graph, type=type, **kwargs)
+        self._preflight_validation(pg_with_metadata=pg_with_metadata)
+        response = self.post(path="/services", json=pg_with_metadata, expected_status=201)
         service_id = response.headers.get("OpenEO-Identifier")
         return Service(service_id, self)
 
@@ -1463,43 +1466,55 @@ def upload_file(
     def _build_request_with_process_graph(self, process_graph: Union[dict, FlatGraphableMixin, Any], **kwargs) -> dict:
         """
         Prepare a json payload with a process graph to submit to /result, /services, /jobs, ...
-        :param process_graph: flat dict representing a process graph
+        :param process_graph: flat dict representing a "process graph with metadata" ({"process": {"process_graph": ...}, ...})
         """
         # TODO: make this a more general helper (like `as_flat_graph`)
         result = kwargs
         process_graph = as_flat_graph(process_graph)
         if "process_graph" not in process_graph:
             process_graph = {"process_graph": process_graph}
-        # TODO: also check if `process_graph` already has "process" key (i.e. is a "process graph with metadata already)
+        # TODO: also check if `process_graph` already has "process" key (i.e. is a "process graph with metadata" already)
         result["process"] = process_graph
         return result
 
-    def _warn_if_process_graph_invalid(self, process_graph: Union[dict, FlatGraphableMixin, str, Path]):
-        # At present, the intention is that a failed validation does not block
-        # the job from running, it is only reported as a warning.
-        # Therefor we also want to continue when something *else* goes wrong
-        # *during* the validation.
-        try:
-            if not self.capabilities().supports_endpoint("/validation", "POST"):
-                return
-
-            graph = self._build_request_with_process_graph(process_graph)["process"]
-            validation_errors = self.validate_process_graph(process_graph=graph)
-            if validation_errors:
-                _log.warning(
-                    "Process graph is not valid. Validation errors:\n"
-                    + "\n".join(e["message"] for e in validation_errors)
-                )
-        except Exception:
-            _log.warning("Could not validate the process graph", exc_info=True)
+    def _preflight_validation(self, pg_with_metadata: dict, *, validate: Optional[bool] = None):
+        """
+        Preflight validation of process graph to execute.
+
+        :param pg_with_metadata: flat dict representation of process graph with metadata,
+            e.g. as produced by `_build_request_with_process_graph`
+        :param validate: Optional toggle to enable/prevent validation of the process graphs before execution
+            (overruling the connection's ``auto_validate`` setting).
+
+        :return:
+        """
+        if validate is None:
+            validate = self._auto_validate
+        if validate and self.capabilities().supports_endpoint("/validation", "POST"):
+            # At present, the intention is that a failed validation does not block
+            # the job from running, it is only reported as a warning.
+            # Therefor we also want to continue when something *else* goes wrong
+            # *during* the validation.
+            try:
+                resp = self.post(path="/validation", json=pg_with_metadata["process"], expected_status=200)
+                validation_errors = resp.json()["errors"]
+                if validation_errors:
+                    _log.warning(
+                        "Preflight process graph validation raised: "
+                        + (" ".join(f"[{e.get('code')}] {e.get('message')}" for e in validation_errors))
+                    )
+            except Exception as e:
+                _log.error(f"Preflight process graph validation failed: {e}", exc_info=True)
+
+        # TODO: additional validation and sanity checks: e.g. is there a result node, are all process_ids valid, ...?
 
     # TODO: unify `download` and `execute` better: e.g. `download` always writes to disk, `execute` returns result (raw or as JSON decoded dict)
     def download(
         self,
         graph: Union[dict, FlatGraphableMixin, str, Path],
         outputfile: Union[Path, str, None] = None,
         timeout: Optional[int] = None,
-        validate: bool = True,
+        validate: Optional[bool] = None,
     ) -> Union[None, bytes]:
         """
         Downloads the result of a process graph synchronously,
@@ -1510,14 +1525,14 @@ def download(
             or as local file path or URL
         :param outputfile: output file
         :param timeout: timeout to wait for response
+        :param validate: Optional toggle to enable/prevent validation of the process graphs before execution
+            (overruling the connection's ``auto_validate`` setting).
         """
-        if validate:
-            self._warn_if_process_graph_invalid(process_graph=graph)
-
-        request = self._build_request_with_process_graph(process_graph=graph)
+        pg_with_metadata = self._build_request_with_process_graph(process_graph=graph)
+        self._preflight_validation(pg_with_metadata=pg_with_metadata, validate=validate)
         response = self.post(
             path="/result",
-            json=request,
+            json=pg_with_metadata,
             expected_status=200,
             stream=True,
             timeout=timeout or DEFAULT_TIMEOUT_SYNCHRONOUS_EXECUTE,
@@ -1534,22 +1549,23 @@ def execute(
         self,
         process_graph: Union[dict, str, Path],
         timeout: Optional[int] = None,
-        validate: bool = True,
+        validate: Optional[bool] = None,
     ):
         """
         Execute a process graph synchronously and return the result (assumed to be JSON).
 
         :param process_graph: (flat) dict representing a process graph, or process graph as raw JSON string,
             or as local file path or URL
+        :param validate: Optional toggle to enable/prevent validation of the process graphs before execution
+            (overruling the connection's ``auto_validate`` setting).
+
         :return: parsed JSON response
         """
-        if validate:
-            self._warn_if_process_graph_invalid(process_graph=process_graph)
-
-        req = self._build_request_with_process_graph(process_graph=process_graph)
+        pg_with_metadata = self._build_request_with_process_graph(process_graph=process_graph)
+        self._preflight_validation(pg_with_metadata=pg_with_metadata, validate=validate)
         return self.post(
             path="/result",
-            json=req,
+            json=pg_with_metadata,
             expected_status=200,
             timeout=timeout or DEFAULT_TIMEOUT_SYNCHRONOUS_EXECUTE,
         ).json()
@@ -1563,7 +1579,7 @@ def create_job(
         plan: Optional[str] = None,
         budget: Optional[float] = None,
         additional: Optional[dict] = None,
-        validate: bool = True,
+        validate: Optional[bool] = None,
     ) -> BatchJob:
         """
         Create a new job from given process graph on the back-end.
@@ -1575,22 +1591,22 @@ def create_job(
         :param plan: billing plan
         :param budget: maximum cost the request is allowed to produce
         :param additional: additional job options to pass to the backend
+        :param validate: Optional toggle to enable/prevent validation of the process graphs before execution
+            (overruling the connection's ``auto_validate`` setting).
         :return: Created job
         """
         # TODO move all this (BatchJob factory) logic to BatchJob?
 
-        if validate:
-            self._warn_if_process_graph_invalid(process_graph=process_graph)
-
-        req = self._build_request_with_process_graph(
+        pg_with_metadata = self._build_request_with_process_graph(
             process_graph=process_graph,
             **dict_no_none(title=title, description=description, plan=plan, budget=budget)
         )
         if additional:
             # TODO: get rid of this non-standard field? https://github.com/Open-EO/openeo-api/issues/276
-            req["job_options"] = additional
+            pg_with_metadata["job_options"] = additional
 
-        response = self.post("/jobs", json=req, expected_status=201)
+        self._preflight_validation(pg_with_metadata=pg_with_metadata, validate=validate)
+        response = self.post("/jobs", json=pg_with_metadata, expected_status=201)
 
         job_id = None
         if "openeo-identifier" in response.headers:
@@ -1668,8 +1684,8 @@ def as_curl(
         cmd += ["-H", "Content-Type: application/json"]
         if isinstance(self.auth, BearerAuth):
             cmd += ["-H", f"Authorization: Bearer {'...' if obfuscate_auth else self.auth.bearer}"]
-        post_data = self._build_request_with_process_graph(data)
-        post_json = json.dumps(post_data, separators=(',', ':'))
+        pg_with_metadata = self._build_request_with_process_graph(data)
+        post_json = json.dumps(pg_with_metadata, separators=(",", ":"))
         cmd += ["--data", post_json]
         cmd += [self.build_url(path)]
         return " ".join(shlex.quote(c) for c in cmd)
@@ -1689,17 +1705,20 @@ def version_info(self):
 
 
 def connect(
-        url: Optional[str] = None,
-        auth_type: Optional[str] = None, auth_options: Optional[dict] = None,
-        session: Optional[requests.Session] = None,
-        default_timeout: Optional[int] = None,
+    url: Optional[str] = None,
+    *,
+    auth_type: Optional[str] = None,
+    auth_options: Optional[dict] = None,
+    session: Optional[requests.Session] = None,
+    default_timeout: Optional[int] = None,
+    auto_validate: bool = True,
 ) -> Connection:
     """
     This method is the entry point to OpenEO.
     You typically create one connection object in your script or application
     and re-use it for all calls to that backend.
 
-    If the backend requires authentication, you can pass authentication data directly to this function
+    If the backend requires authentication, you can pass authentication data directly to this function,
     but it could be easier to authenticate as follows:
 
         >>> # For basic authentication
@@ -1711,7 +1730,10 @@ def connect(
     :param auth_type: Which authentication to use: None, "basic" or "oidc" (for OpenID Connect)
     :param auth_options: Options/arguments specific to the authentication type
     :param default_timeout: default timeout (in seconds) for requests
-    :rtype: openeo.connections.Connection
+    :param auto_validate: toggle to automatically validate process graphs before execution
+
+    .. versionadded:: 0.24.0
+        added ``auto_validate`` argument
     """
 
     def _config_log(message):
@@ -1736,7 +1758,7 @@ 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)
+    connection = Connection(url, session=session, default_timeout=default_timeout, auto_validate=auto_validate)
 
     auth_type = auth_type.lower() if isinstance(auth_type, str) else auth_type
     if auth_type in {None, False, 'null', 'none'}:

openeo/rest/datacube.py

@@ -1981,7 +1981,8 @@ def download(
         outputfile: Optional[Union[str, pathlib.Path]] = None,
         format: Optional[str] = None,
         options: Optional[dict] = None,
-        validate: bool = True,
+        *,
+        validate: Optional[bool] = None,
     ) -> Union[None, bytes]:
         """
         Execute synchronously and download the raster data cube, e.g. as GeoTIFF.
@@ -1992,6 +1993,8 @@ def download(
         :param outputfile: Optional, an output file if the result needs to be stored on disk.
         :param format: Optional, an output format supported by the backend.
         :param options: Optional, file format options
+        :param validate: Optional toggle to enable/prevent validation of the process graphs before execution
+            (overruling the connection's ``auto_validate`` setting).
         :return: None if the result is stored to disk, or a bytes object returned by the backend.
         """
         if format is None and outputfile:
@@ -2099,7 +2102,7 @@ def execute_batch(
         max_poll_interval: float = 60,
         connection_retry_interval: float = 30,
         job_options: Optional[dict] = None,
-        validate: bool = True,
+        validate: Optional[bool] = None,
         # TODO: avoid `format_options` as keyword arguments
         **format_options,
     ) -> BatchJob:
@@ -2112,6 +2115,8 @@ def execute_batch(
         :param outputfile: The path of a file to which a result can be written
         :param out_format: (optional) File format to use for the job result.
         :param job_options:
+        :param validate: Optional toggle to enable/prevent validation of the process graphs before execution
+            (overruling the connection's ``auto_validate`` setting).
         """
         if "format" in format_options and not out_format:
             out_format = format_options["format"]  # align with 'download' call arg name
@@ -2134,7 +2139,7 @@ def create_job(
         plan: Optional[str] = None,
         budget: Optional[float] = None,
         job_options: Optional[dict] = None,
-        validate: bool = True,
+        validate: Optional[bool] = None,
         # TODO: avoid `format_options` as keyword arguments
         **format_options,
     ) -> BatchJob:
@@ -2152,6 +2157,9 @@ def create_job(
         :param plan: billing plan
         :param budget: maximum cost the request is allowed to produce
         :param job_options: custom job options.
+        :param validate: Optional toggle to enable/prevent validation of the process graphs before execution
+            (overruling the connection's ``auto_validate`` setting).
+
         :return: Created job.
         """
         # TODO: add option to also automatically start the job?
@@ -2200,8 +2208,8 @@ def save_user_defined_process(
             returns=returns, categories=categories, examples=examples, links=links,
         )
 
-    def execute(self, validate: bool = True) -> dict:
-        """Executes the process graph of the imagery. """
+    def execute(self, *, validate: Optional[bool] = None) -> dict:
+        """Executes the process graph."""
         return self._connection.execute(self.flat_graph(), validate=validate)
 
     @staticmethod

openeo/rest/udp.py

@@ -94,6 +94,7 @@ def store(
         # TODO: this "public" flag is not standardized yet EP-3609, https://github.com/Open-EO/openeo-api/issues/310
         process["public"] = public
 
+        self._connection._preflight_validation(pg_with_metadata=process)
         self._connection.put(
             path="/process_graphs/{}".format(self.user_defined_process_id), json=process, expected_status=200
         )