Merge branch 'issue744_download_optional'

Author: soxofaan

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `MultiBackendJobManager`: add `download_results` option to enable/disable the automated download of job results once completed by the job manager ([#744](https://github.com/Open-EO/openeo-python-client/issues/744))
+
 ### Changed
 
 - Internal reorganization of `openeo.extra.job_management` submodule to ease future development ([#741](https://github.com/Open-EO/openeo-python-client/issues/741))

openeo/extra/job_management/_manager.py

@@ -157,14 +157,21 @@ def start_job(
             - get_error_log_path
             - get_job_metadata_path
 
+    :param download_results:
+        Whether to download job results automatically once the job is completed.
+
     :param cancel_running_job_after:
         Optional temporal limit (in seconds) after which running jobs should be canceled
         by the job manager.
 
+
     .. versionadded:: 0.14.0
 
     .. versionchanged:: 0.32.0
         Added ``cancel_running_job_after`` parameter.
+
+    .. versionchanged:: 0.47.0
+        Added ``download_results`` parameter.
     """
 
     # Expected columns in the job DB dataframes.
@@ -193,6 +200,7 @@ def __init__(
         poll_sleep: int = 60,
         root_dir: Optional[Union[str, Path]] = ".",
         *,
+        download_results: bool = True,
         cancel_running_job_after: Optional[int] = None,
     ):
         """Create a MultiBackendJobManager."""
@@ -204,6 +212,8 @@ def __init__(
         # An explicit None or "" should also default to "."
         self._root_dir = Path(root_dir or ".")
 
+        self._download_results = download_results
+
         self._cancel_running_job_after = (
             datetime.timedelta(seconds=cancel_running_job_after) if cancel_running_job_after is not None else None
         )
@@ -712,16 +722,16 @@ def on_job_done(self, job: BatchJob, row):
         :param row: DataFrame row containing the job's metadata.
         """
         # TODO: param `row` is never accessed in this method. Remove it? Is this intended for future use?
+        if self._download_results:
+            job_metadata = job.describe()
+            job_dir = self.get_job_dir(job.job_id)
+            metadata_path = self.get_job_metadata_path(job.job_id)
 
-        job_metadata = job.describe()
-        job_dir = self.get_job_dir(job.job_id)
-        metadata_path = self.get_job_metadata_path(job.job_id)
-
-        self.ensure_job_dir_exists(job.job_id)
-        job.get_results().download_files(target=job_dir)
+            self.ensure_job_dir_exists(job.job_id)
+            job.get_results().download_files(target=job_dir)
 
-        with metadata_path.open("w", encoding="utf-8") as f:
-            json.dump(job_metadata, f, ensure_ascii=False)
+            with metadata_path.open("w", encoding="utf-8") as f:
+                json.dump(job_metadata, f, ensure_ascii=False)
 
     def on_job_error(self, job: BatchJob, row):
         """

tests/extra/job_management/test_manager.py

@@ -39,7 +39,7 @@
 )
 from openeo.rest._testing import DummyBackend
 from openeo.rest.auth.testing import OidcMock
-from openeo.util import rfc3339
+from openeo.util import load_json, rfc3339
 
 
 def _job_id_from_year(process_graph) -> Union[str, None]:
@@ -897,3 +897,33 @@ def test_refresh_bearer_token_before_start(
         # Because of proactive+throttled token refreshing,
         # we should have 2 additional token requests now
         assert len(oidc_mock.grant_request_history) == 4
+
+    @pytest.mark.parametrize(
+        ["download_results"],
+        [
+            (True,),
+            (False,),
+        ],
+    )
+    def test_download_results_toggle(
+        self, tmp_path, job_manager_root_dir, dummy_backend_foo, download_results, sleep_mock
+    ):
+        job_manager = MultiBackendJobManager(root_dir=job_manager_root_dir, download_results=download_results)
+        job_manager.add_backend("foo", connection=dummy_backend_foo.connection)
+
+        df = pd.DataFrame({"year": [2018, 2019]})
+        job_db = CsvJobDatabase(tmp_path / "jobs.csv").initialize_from_df(df)
+        run_stats = job_manager.run_jobs(job_db=job_db, start_job=self._create_year_job)
+        assert run_stats == dirty_equals.IsPartialDict({"job finished": 2})
+
+        if download_results:
+            assert (job_manager_root_dir / "job_job-2018/result.data").read_bytes() == DummyBackend.DEFAULT_RESULT
+            assert load_json(job_manager_root_dir / "job_job-2018/job_job-2018.json") == dirty_equals.IsPartialDict(
+                id="job-2018", status="finished"
+            )
+            assert (job_manager_root_dir / "job_job-2019/result.data").read_bytes() == DummyBackend.DEFAULT_RESULT
+            assert load_json(job_manager_root_dir / "job_job-2019/job_job-2019.json") == dirty_equals.IsPartialDict(
+                id="job-2019", status="finished"
+            )
+        else:
+            assert not job_manager_root_dir.exists() or list(job_manager_root_dir.iterdir()) == []