Merge branch 'issue195-apiv12-processesv20'

Author: soxofaan

CHANGELOG.md

@@ -21,6 +21,15 @@ and start a new "In Progress" section above it.
 
 ## In progress
 
+- Add STAC collections conformance class ([#195](https://github.com/Open-EO/openeo-python-driver/issues/195))
+- update openeo_driver/specs/openeo-api/1.x submodule to tag `1.2.0` ([#195](https://github.com/Open-EO/openeo-python-driver/issues/195))
+
+
+## 0.125.0
+
+- Add log level to batch job logs response ([#195](https://github.com/Open-EO/openeo-python-driver/issues/195))
+
+
 ## 0.124.0
 
 - Better argument validation in `resample_spatial`/`resample_cube_spatial` (related to [Open-EO/openeo-python-client#690](https://github.com/Open-EO/openeo-python-client/issues/690))

openeo_driver/_version.py

@@ -1 +1 @@
-__version__ = "0.124.0a1"
+__version__ = "0.125.0a1"

openeo_driver/backend.py

@@ -31,7 +31,7 @@
 from openeo_driver.datastructs import SarBackscatterArgs
 from openeo_driver.dry_run import SourceConstraint
 from openeo_driver.errors import CollectionNotFoundException, ServiceUnsupportedException, FeatureUnsupportedException
-from openeo_driver.constants import JOB_STATUS
+from openeo_driver.constants import JOB_STATUS, DEFAULT_LOG_LEVEL_RETRIEVAL
 from openeo_driver.processes import ProcessRegistry
 from openeo_driver.users import User
 from openeo_driver.users.oidc import OidcProvider
@@ -142,7 +142,15 @@ def remove_service(self, user_id: str, service_id: str) -> None:
         """https://openeo.org/documentation/1.0/developers/api/reference.html#operation/delete-service"""
         raise NotImplementedError()
 
-    def get_log_entries(self, service_id: str, user_id: str, offset: str) -> List[dict]:
+    def get_log_entries(
+        self,
+        service_id: str,
+        *,
+        user_id: str,
+        offset: Union[str, None] = None,
+        limit: Union[int, None] = None,
+        level: str = DEFAULT_LOG_LEVEL_RETRIEVAL,
+    ) -> List[dict]:
         """https://openeo.org/documentation/1.0/developers/api/reference.html#operation/debug-service"""
         # TODO require auth/user handle?
         return []
@@ -178,12 +186,12 @@ class LoadParameters(dict):
     """
     A buffer provided in the units of the target CRS. If target CRS is not provided, then it is assumed to be the native CRS
     of the collection.
-    
-    This buffer is applied to AOI when constructing the datacube, allowing operations that require neighbouring pixels 
+
+    This buffer is applied to AOI when constructing the datacube, allowing operations that require neighbouring pixels
     to be implemented correctly. Examples are apply_kernel and apply_neighborhood, but also certain resampling operations
     could be affected by this.
-    
-    The buffer has to be considered in the global extent! 
+
+    The buffer has to be considered in the global extent!
     """
     pixel_buffer = dict_item(default=None)
 
@@ -552,9 +560,11 @@ def get_results(self, job_id: str, user_id: str) -> Dict[str, dict]:
     def get_log_entries(
         self,
         job_id: str,
+        *,
         user_id: str,
-        offset: Optional[str] = None,
-        level: Optional[str] = None,
+        offset: Union[str, None] = None,
+        limit: Union[str, None] = None,
+        level: str = DEFAULT_LOG_LEVEL_RETRIEVAL,
     ) -> Iterable[dict]:
         """
         https://openeo.org/documentation/1.0/developers/api/reference.html#operation/debug-job
@@ -760,6 +770,9 @@ class OpenEoBackendImplementation:
         "https://api.openeo.org/1.2.0",
         # Support the "remote process definition" extension (originally known as the "remote-udp" extension)
         "https://api.openeo.org/extensions/remote-process-definition/0.1.0",
+        # STAC API conformance classes
+        # "https://api.stacspec.org/v1.0.0/core",  # TODO #363 can we claim this conformance class already?
+        "https://api.stacspec.org/v1.0.0/collections",
     ]
 
     def __init__(

openeo_driver/constants.py

@@ -49,3 +49,9 @@ class JOB_STATUS:
     "lower-right",
     "upper-right",
 ]
+
+
+# Default value for `level` parameter in `POST /result`, `POST /jobs`, ... requests
+DEFAULT_LOG_LEVEL_PROCESSING = "info"
+# Default value for `level in `GET /jobs/{job_id}/logs`, `GET /services/{service_id}/logs` requests
+DEFAULT_LOG_LEVEL_RETRIEVAL = "debug"

openeo_driver/dummy/dummy_backend.py

@@ -43,7 +43,7 @@
     UserDefinedProcessMetadata,
 )
 from openeo_driver.config import OpenEoBackendConfig
-from openeo_driver.constants import JOB_STATUS, STAC_EXTENSION
+from openeo_driver.constants import JOB_STATUS, STAC_EXTENSION, DEFAULT_LOG_LEVEL_RETRIEVAL
 from openeo_driver.datacube import DriverDataCube, DriverMlModel, DriverVectorCube
 from openeo_driver.datastructs import StacAsset
 from openeo_driver.delayed_vector import DelayedVector
@@ -172,9 +172,17 @@ def list_services(self, user_id: str) -> List[ServiceMetadata]:
     def service_info(self, user_id: str, service_id: str) -> ServiceMetadata:
         return next(s for s in self._registry if s.id == service_id)
 
-    def get_log_entries(self, service_id: str, user_id: str, offset: str) -> List[dict]:
+    def get_log_entries(
+        self,
+        service_id: str,
+        *,
+        user_id: str,
+        offset: Union[str, None] = None,
+        limit: Union[int, None] = None,
+        level: str = DEFAULT_LOG_LEVEL_RETRIEVAL,
+    ) -> List[dict]:
         return [
-            {"id": 3, "level": "info", "message": "Loaded data."}
+            {"id": 3, "level": "info", "message": "Loaded data."},
         ]
 
 
@@ -921,16 +929,18 @@ def get_result_assets(self, job_id: str, user_id: str) -> Dict[str, dict]:
     def get_log_entries(
         self,
         job_id: str,
+        *,
         user_id: str,
-        offset: Optional[str] = None,
-        level: Optional[str] = None,
+        offset: Union[str, None] = None,
+        limit: Union[str, None] = None,
+        level: str = DEFAULT_LOG_LEVEL_RETRIEVAL,
     ) -> Iterable[dict]:
         self._get_job_info(job_id=job_id, user_id=user_id)
         default_logs = [{"id": "1", "level": "info", "message": "hello world"}]
         requested_level = normalize_log_level(level)
         for log in self._custom_job_logs.get(job_id, default_logs):
             if isinstance(log, dict):
-                actual_level = normalize_log_level(log.get("log_level"))
+                actual_level = normalize_log_level(log.get("level"))
                 if actual_level < requested_level:
                     continue
                 yield log

openeo_driver/errors.py

@@ -360,7 +360,7 @@ class ProcessGraphMissingException(OpenEOApiException):
     status_code = 400
     code = 'ProcessGraphMissing'
     message = "Invalid process specified. It doesn't contain a process graph."
-    _description = "The parameter `process` doesn't contain a valid process."
+    _description = "The process doesn't contain a process graph. For jobs, services, and sync. processing the parameter `process` must contain a `process_graph`."
     _tags = ['Batch Jobs', 'Data Processing', 'Secondary Services', 'User-Defined Processes']
 
 
@@ -442,19 +442,19 @@ class ProcessUnsupportedException(OpenEOApiException):
     status_code = 400
     code = 'ProcessUnsupported'
     message = "Process with identifier '{process}' is not available in namespace '{namespace}'."
-    _description = 'A process (pre-defined or user-defined) with the specified identifier is not available. To be used when validating or executing process graphs.'
-    _tags = ['Data Processing']
+    _description = "A process (predefined or user-defined) with the specified identifier is not available. To be used when validating or executing process graphs."
+    _tags = ["Data Processing"]
 
     def __init__(self, process: str, namespace: str = "backend"):
         super().__init__(message=self.message.format(process=process, namespace=namespace))
 
 
 class PredefinedProcessExistsException(OpenEOApiException):
     status_code = 400
-    code = 'PredefinedProcessExists'
-    message = 'A predefined process with the given identifier exists.'
-    _description = 'If a user wants to store a user-defined process with the id of a pre-defined process.'
-    _tags = ['User-Defined Processes']
+    code = "PredefinedProcessExists"
+    message = "A predefined process with the given identifier exists."
+    _description = "If a user wants to store a user-defined process with the id of a predefined process."
+    _tags = ["User-Defined Processes"]
 
 
 class ProcessParameterRequiredException(OpenEOApiException):
@@ -542,6 +542,24 @@ def __init__(self, parameter: str, reason: str):
         super().__init__(message=self.message.format(parameter=parameter, reason=reason))
 
 
+class EstimateComplexityException(OpenEOApiException):
+    status_code = 500
+    code = "EstimateComplexity"
+    message = "The process is too complex to calculate an estimate."
+    _description = "The process is too complex to calculate an estimate, e.g. due to a UDF or other processes that are complex to estimate costs reliably."
+    _tags = ["Batch Jobs"]
+
+
+class BillingPlanMissingException(OpenEOApiException):
+    status_code = 400
+    code = "BillingPlanMissing"
+    message = "A billing plan must be specified."
+    _description = (
+        "No billing plan has been specified by the user and the billing plan can't be determined unambiguously."
+    )
+    _tags = ["Batch Jobs", "Data Processing", "Secondary Services"]
+
+
 # --- End of semi-autogenerated openEO exception classes ------------------------------------------------
 
 

openeo_driver/specs/openeo-api/1.x

@@ -206,9 +206,11 @@ def ensure_auth_header(self):
         if not self.default_request_headers.get("Authorization"):
             self.set_auth_bearer_token()
 
-    def get(self, path: str, headers: dict = None) -> ApiResponse:
+    def get(self, path: str, headers: dict = None, params: Optional[dict] = None) -> ApiResponse:
         """Do versioned GET request, given non-versioned path"""
-        return ApiResponse(self.client.get(path=self.url(path), headers=self._request_headers(headers)))
+        return ApiResponse(
+            self.client.get(path=self.url(path), headers=self._request_headers(headers), query_string=params)
+        )
 
     def head(self, path: str, headers: dict = None) -> ApiResponse:
         """Do versioned GET request, given non-versioned path"""

openeo_driver/testing.py

@@ -39,7 +39,7 @@
     JobListing,
 )
 from openeo_driver.config import get_backend_config, OpenEoBackendConfig
-from openeo_driver.constants import STAC_EXTENSION
+from openeo_driver.constants import STAC_EXTENSION, DEFAULT_LOG_LEVEL_RETRIEVAL, DEFAULT_LOG_LEVEL_PROCESSING
 from openeo_driver.datacube import DriverMlModel
 from openeo_driver.errors import (
     OpenEOApiException,
@@ -404,7 +404,8 @@ def index():
             "version": api_version,  # Deprecated pre-0.4.0 API version field
             "api_version": api_version,  # API version field since 0.4.0
             "backend_version": backend_version,
-            "stac_version": "0.9.0",
+            "stac_version": "0.9.0",  # TODO #363 bump to 1.x.y?
+            "type": "Catalog",
             "conformsTo": backend_implementation.conformance_classes(),
             "id": service_id,
             "title": title,
@@ -654,7 +655,7 @@ def result(user: User):
         process_graph = _extract_process_graph(post_data)
         budget = post_data.get("budget")
         plan = post_data.get("plan")
-        log_level = post_data.get("log_level", "info")
+        log_level = post_data.get("log_level", DEFAULT_LOG_LEVEL_PROCESSING)
         job_options = _extract_job_options(
             post_data, to_ignore=["process", "process_graph", "budget", "plan", "log_level"]
         )
@@ -880,7 +881,7 @@ def create_job(user: User):
             post_data, to_ignore=["process", "process_graph", "title", "description", "plan", "budget", "log_level"]
         )
         metadata = {k: post_data[k] for k in ["title", "description", "plan", "budget"] if k in post_data}
-        metadata["log_level"] = post_data.get("log_level", "info")
+        metadata["log_level"] = post_data.get("log_level", DEFAULT_LOG_LEVEL_PROCESSING)
         job_info = backend_implementation.batch_jobs.create_job(
             # TODO: remove `filter_supported_kwargs` (when all implementations have migrated to `user` iso `user_id`)
             **filter_supported_kwargs(
@@ -1513,16 +1514,26 @@ def download_job_result_signed(job_id, user_base64, secure_key, filename):
     @blueprint.route("/jobs/<job_id>/logs", methods=["GET"])
     @auth_handler.requires_bearer_auth
     def get_job_logs(job_id, user: User):
-        offset = request.args.get("offset")
-        level = request.args.get("level", "debug")
+        offset = request.args.get("offset", default=None)
+        limit = request.args.get("limit", default=None, type=int)
+        level = request.args.get("level", default=DEFAULT_LOG_LEVEL_RETRIEVAL)
         request_id = FlaskRequestCorrelationIdLogging.get_request_id()
         # TODO: implement paging support: `limit`, next/prev/first/last `links`, ...
-        logs = backend_implementation.batch_jobs.get_log_entries(
+
+        # TODO: remove this `function_has_argument` once all implementations are migrated
+        if function_has_argument(backend_implementation.batch_jobs.get_log_entries, argument="limit"):
+            logs = backend_implementation.batch_jobs.get_log_entries(
+                job_id=job_id, user_id=user.user_id, offset=offset, level=level, limit=limit
+            )
+        else:
+            logs = backend_implementation.batch_jobs.get_log_entries(
             job_id=job_id, user_id=user.user_id, offset=offset, level=level
         )
 
         def generate():
-            yield """{"logs":["""
+            yield "{"
+            yield f'"level": {json.dumps(level)},'
+            yield '"logs":['
 
             sep = ""
             try:
@@ -1550,7 +1561,9 @@ def generate():
                 }
                 yield sep + json.dumps(log)
 
-            yield """],"links":[]}"""
+            yield "],"
+            # TODO: add pagination links (next, prev, first, last)
+            yield '"links":[]}'
 
         return current_app.response_class(generate(), mimetype="application/json")
 
@@ -1665,12 +1678,13 @@ def service_delete(service_id, user: User):
     @blueprint.route('/services/<service_id>/logs', methods=['GET'])
     @auth_handler.requires_bearer_auth
     def service_logs(service_id, user: User):
-        level = request.args.get("level", "debug")
-        offset = request.args.get('offset', 0)
+        offset = request.args.get("offset", default=None)
+        limit = request.args.get("limit", default=None, type=int)
+        level = request.args.get("level", default=DEFAULT_LOG_LEVEL_RETRIEVAL)
         logs = backend_implementation.secondary_services.get_log_entries(
-            service_id=service_id, user_id=user.user_id, offset=offset
+            service_id=service_id, user_id=user.user_id, offset=offset, limit=limit, level=level
         )
-        return jsonify({"logs": logs, "links": []})
+        return jsonify({"logs": logs, "links": [], "level": level})
 
 
 def register_views_udp(

openeo_driver/views.py

@@ -84,6 +84,8 @@ def test_error_code(error_code):
             raise Exception("Exception class {c} has placeholder in message but no custom __init__".format(
                 c=exception_cls.__name__))
 
+        assert exception_cls._description == error_spec["description"]
+
 
 def test_flask_request_id_as_api_error_id():
     app = flask.Flask(__name__)