Merge branch 'master' of https://github.com/Open-EO/openeo-python-client into issue505-Option_to_disable_printing_error_logs_on_failed_job

# Conflicts:
#	CHANGELOG.md

Author: ElienVandermaesenVITO

CHANGELOG.md

@@ -9,8 +9,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+### Changed
+
+### Removed
+
+### Fixed
+
+
+## [0.36.0] - 2024-12-10
+
+### Added
+
 - Automatically use `load_url` when providing a URL as geometries to `DataCube.aggregate_spatial()`, `DataCube.mask_polygon()`, etc. ([#104](https://github.com/Open-EO/openeo-python-client/issues/104), [#457](https://github.com/Open-EO/openeo-python-client/issues/457))
-- Argument `spatial_extent` in `load_collection` supports Shapely objects and loading GeoJSON from a local path.
+- Allow specifying `limit` when listing batch jobs with `Connection.list_jobs()` ([#677](https://github.com/Open-EO/openeo-python-client/issues/677))
+- Add `additional` and `job_options` arguments to `Connection.download()`, `Datacube.download()` and related ([#681](https://github.com/Open-EO/openeo-python-client/issues/681))
 
 ### Changed
 
@@ -19,12 +31,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   this is not translated automatically anymore to deprecated, non-standard `read_vector` usage.
   Instead, if it is a local GeoJSON file, the GeoJSON data will be loaded directly client-side.
   ([#104](https://github.com/Open-EO/openeo-python-client/issues/104), [#457](https://github.com/Open-EO/openeo-python-client/issues/457))
-
-### Removed
+- Move `read()` method from general `JobDatabaseInterface` to more specific `FullDataFrameJobDatabase` ([#680](https://github.com/Open-EO/openeo-python-client/issues/680))
+- Align `additional` and `job_options` arguments in `Connection.create_job()`, `DataCube.create_job()` and related.
+  Also, follow official spec more closely. ([#683](https://github.com/Open-EO/openeo-python-client/issues/683), [Open-EO/openeo-api#276](https://github.com/Open-EO/openeo-api/issues/276))
 
 ### Fixed
 
 - `load_stac`: use fallback temporal dimension when no "cube:dimensions" in STAC Collection ([#666](https://github.com/Open-EO/openeo-python-client/issues/666))
+- Fix usage of `Parameter.spatial_extent()` with `load_collection` and `filter_bbox` ([#676](https://github.com/Open-EO/openeo-python-client/issues/676))
+
 
 ## [0.35.0] - 2024-11-19
 

docs/udp.rst

@@ -134,7 +134,7 @@ Some useful parameter helpers (class methods of the :py:class:`~openeo.api.proce
 -   :py:meth:`Parameter.geojson() <openeo.api.process.Parameter.geojson>` to create
     a parameter for specifying a GeoJSON geometry.
 -   :py:meth:`Parameter.spatial_extent() <openeo.api.process.Parameter.spatial_extent>` to create
-    a spatial_extent parameter that is exactly the same as the corresponding parameter in `load_collection` and `load_stac`.
+    a spatial_extent parameter that is exactly the same as the corresponding parameter in ``load_collection`` and ``load_stac``.
 
 
 

openeo/_version.py

@@ -1 +1 @@
-__version__ = "0.36.0a1"
+__version__ = "0.37.0a1"

openeo/api/process.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import textwrap
 import warnings
 from typing import List, Optional, Union
 
@@ -279,23 +280,15 @@ def bounding_box(
         }
         return cls(name=name, description=description, schema=schema, **kwargs)
 
-    _spatial_extent_description = """Limits the data to process to the specified bounding box or polygons.
-
-For raster data, the process loads the pixel into the data cube if the point at the pixel center intersects with the bounding box or any of the polygons (as defined in the Simple Features standard by the OGC).
-For vector data, the process loads the geometry into the data cube if the geometry is fully within the bounding box or any of the polygons (as defined in the Simple Features standard by the OGC). Empty geometries may only be in the data cube if no spatial extent has been provided.
-
-Empty geometries are ignored.
-Set this parameter to null to set no limit for the spatial extent. """
-
     @classmethod
     def spatial_extent(
         cls,
         name: str = "spatial_extent",
-        description: str = _spatial_extent_description,
+        description: Optional[str] = None,
         **kwargs,
     ) -> Parameter:
         """
-        Helper to easily create a 'spatial_extent' parameter, which is compatible with the 'load_collection' argument of
+        Helper to easily create a 'spatial_extent' parameter, which is compatible with the ``load_collection`` argument of
         the same name. This allows to conveniently create user-defined processes that can be applied to a bounding box and vector data
         for spatial filtering. It is also possible for users to set to null, and define spatial filtering using other processes.
 
@@ -307,6 +300,26 @@ def spatial_extent(
 
         .. versionadded:: 0.32.0
         """
+        if description is None:
+            description = textwrap.dedent(
+                """
+                Limits the data to process to the specified bounding box or polygons.
+
+                For raster data, the process loads the pixel into the data cube if the point
+                at the pixel center intersects with the bounding box or any of the polygons
+                (as defined in the Simple Features standard by the OGC).
+
+                For vector data, the process loads the geometry into the data cube if the geometry
+                is fully within the bounding box or any of the polygons (as defined in the
+                Simple Features standard by the OGC). Empty geometries may only be in the
+                data cube if no spatial extent has been provided.
+
+                Empty geometries are ignored.
+
+                Set this parameter to null to set no limit for the spatial extent.
+                """
+            ).strip()
+
         schema = [
             {
                 "title": "Bounding Box",
@@ -410,7 +423,7 @@ def geojson(cls, name: str, description: str = "Geometries specified as GeoJSON
     @classmethod
     def temporal_interval(
         cls,
-        name: str,
+        name: str = "temporal_extent",
         description: str = "Temporal extent specified as two-element array with start and end date/date-time.",
         **kwargs,
     ) -> Parameter:
@@ -441,3 +454,26 @@ def temporal_interval(
             },
         }
         return cls(name=name, description=description, schema=schema, **kwargs)
+
+
+def schema_supports(schema: Union[dict, List[dict]], type: str, subtype: Optional[str] = None) -> bool:
+    """Helper to check if parameter schema supports given type/subtype"""
+    # TODO: support checking item type in arrays
+    if isinstance(schema, dict):
+        actual_type = schema.get("type")
+        if isinstance(actual_type, str):
+            if actual_type != type:
+                return False
+        elif isinstance(actual_type, list):
+            if type not in actual_type:
+                return False
+        else:
+            raise ValueError(actual_type)
+        if subtype:
+            if schema.get("subtype") != subtype:
+                return False
+        return True
+    elif isinstance(schema, list):
+        return any(schema_supports(s, type=type, subtype=subtype) for s in schema)
+    else:
+        raise ValueError(schema)

openeo/extra/job_management.py openeo/extra/job_management/__init__.pyopeneo/extra/job_management.py renamed to openeo/extra/job_management/__init__.py

@@ -41,6 +41,7 @@
 
 _log = logging.getLogger(__name__)
 
+
 class _Backend(NamedTuple):
     """Container for backend info/settings"""
 
@@ -70,15 +71,6 @@ def exists(self) -> bool:
         """Does the job database already exist, to read job data from?"""
         ...
 
-    @abc.abstractmethod
-    def read(self) -> pd.DataFrame:
-        """
-        Read job data from the database as pandas DataFrame.
-
-        :return: loaded job data.
-        """
-        ...
-
     @abc.abstractmethod
     def persist(self, df: pd.DataFrame):
         """
@@ -364,9 +356,9 @@ def start_job_thread(self, start_job: Callable[[], BatchJob], job_db: JobDatabas
 
         # Resume from existing db
         _log.info(f"Resuming `run_jobs` from existing {job_db}")
-        df = job_db.read()
 
         self._stop_thread = False
+
         def run_loop():
 
             # TODO: support user-provided `stats`
@@ -810,6 +802,15 @@ def initialize_from_df(self, df: pd.DataFrame, *, on_exists: str = "error"):
         # Return self to allow chaining with constructor.
         return self
 
+    @abc.abstractmethod
+    def read(self) -> pd.DataFrame:
+        """
+        Read job data from the database as pandas DataFrame.
+
+        :return: loaded job data.
+        """
+        ...
+
     @property
     def df(self) -> pd.DataFrame:
         if self._df is None:
@@ -856,6 +857,7 @@ class CsvJobDatabase(FullDataFrameJobDatabase):
 
     .. versionadded:: 0.31.0
     """
+
     def __init__(self, path: Union[str, Path]):
         super().__init__()
         self.path = Path(path)
@@ -912,6 +914,7 @@ class ParquetJobDatabase(FullDataFrameJobDatabase):
 
     .. versionadded:: 0.31.0
     """
+
     def __init__(self, path: Union[str, Path]):
         super().__init__()
         self.path = Path(path)
@@ -934,6 +937,7 @@ def read(self) -> pd.DataFrame:
         metadata = pyarrow.parquet.read_metadata(self.path)
         if b"geo" in metadata.metadata:
             import geopandas
+
             return geopandas.read_parquet(self.path)
         else:
             return pd.read_parquet(self.path)
@@ -1045,6 +1049,7 @@ class ProcessBasedJobCreator:
         `feedback and suggestions for improvement <https://github.com/Open-EO/openeo-python-client/issues>`_.
 
     """
+
     def __init__(
         self,
         *,
@@ -1077,7 +1082,6 @@ def _get_process_definition(self, connection: Connection) -> Process:
                 f"Unsupported process definition source udp_id={self._process_id!r} namespace={self._namespace!r}"
             )
 
-
     def start_job(self, row: pd.Series, connection: Connection, **_) -> BatchJob:
         """
         Implementation of the ``start_job`` callable interface

openeo/rest/_testing.py

@@ -32,13 +32,17 @@ class DummyBackend:
     """
 
     # TODO: move to openeo.testing
+    # TODO: unify "batch_jobs", "batch_jobs_full" and "extra_job_metadata_fields"?
+    # TODO: unify "sync_requests" and "sync_requests_full"?
 
     __slots__ = (
         "_requests_mock",
         "connection",
         "file_formats",
         "sync_requests",
+        "sync_requests_full",
         "batch_jobs",
+        "batch_jobs_full",
         "validation_requests",
         "next_result",
         "next_validation_errors",
@@ -60,7 +64,9 @@ def __init__(
         self.connection = connection
         self.file_formats = {"input": {}, "output": {}}
         self.sync_requests = []
+        self.sync_requests_full = []
         self.batch_jobs = {}
+        self.batch_jobs_full = {}
         self.validation_requests = []
         self.next_result = self.DEFAULT_RESULT
         self.next_validation_errors = []
@@ -163,7 +169,9 @@ def setup_file_format(self, name: str, type: str = "output", gis_data_types: Ite
 
     def _handle_post_result(self, request, context):
         """handler of `POST /result` (synchronous execute)"""
-        pg = request.json()["process"]["process_graph"]
+        post_data = request.json()
+        pg = post_data["process"]["process_graph"]
+        self.sync_requests_full.append(post_data)
         self.sync_requests.append(pg)
         result = self.next_result
         if isinstance(result, (dict, list)):
@@ -185,6 +193,10 @@ def _handle_post_jobs(self, request, context):
             job_id = f"job-{len(self.batch_jobs):03d}"
         assert job_id not in self.batch_jobs
 
+        # Full post data dump
+        self.batch_jobs_full[job_id] = post_data
+
+        # Batch job essentials
         job_data = {"job_id": job_id, "pg": pg, "status": "created"}
         for field in ["title", "description"]:
             if field in post_data:
@@ -272,6 +284,11 @@ def get_sync_pg(self) -> dict:
         assert len(self.sync_requests) == 1
         return self.sync_requests[0]
 
+    def get_sync_post_data(self) -> dict:
+        """Get post data of the one and only synchronous job."""
+        assert len(self.sync_requests_full) == 1
+        return self.sync_requests_full[0]
+
     def get_batch_pg(self) -> dict:
         """
         Get process graph of the one and only batch job.
@@ -280,6 +297,14 @@ def get_batch_pg(self) -> dict:
         assert len(self.batch_jobs) == 1
         return self.batch_jobs[max(self.batch_jobs.keys())]["pg"]
 
+    def get_batch_post_data(self) -> dict:
+        """
+        Get post data of the one and only batch job.
+        Fails when there is none or more than one.
+        """
+        assert len(self.batch_jobs_full) == 1
+        return self.batch_jobs_full[max(self.batch_jobs_full.keys())]
+
     def get_validation_pg(self) -> dict:
         """
         Get process graph of the one and only validation request.