fixup! PR #736 addressing some review notes (exclude bearer from repr)

Author: soxofaan

openeo/extra/job_management/__init__.py

@@ -683,11 +683,13 @@ def _process_threadworker_updates(
             # Process database updates
             if res.db_update:
                 try:
-                    updates.append({
-                        'id': res.job_id,
-                        'df_idx': res.df_idx,
-                        **res.db_update,
-                    })
+                    updates.append(
+                        {
+                            "id": res.job_id,
+                            "df_idx": res.df_idx,
+                            **res.db_update,
+                        }
+                    )
                 except Exception as e:
                     _log.error(f"Skipping invalid db_update {res.db_update!r} for job {res.job_id!r}: {e}")
 
@@ -705,21 +707,20 @@ def _process_threadworker_updates(
             return
 
         # Build DataFrame of updates indexed by df_idx
-        df_updates = pd.DataFrame(updates).set_index('df_idx', drop=True)
+        df_updates = pd.DataFrame(updates).set_index("df_idx", drop=True)
 
         # Determine which rows to upsert
         existing_indices = set(df_updates.index).intersection(job_db.read().index)
         if existing_indices:
             df_upsert = df_updates.loc[sorted(existing_indices)]
             job_db.persist(df_upsert)
-            stats['job_db persist'] = stats.get('job_db persist', 0) + 1
+            stats["job_db persist"] = stats.get("job_db persist", 0) + 1
 
         # Any df_idx not in original index are errors
         missing = set(df_updates.index) - existing_indices
         if missing:
             _log.error(f"Skipping non-existing dataframe indices: {sorted(missing)}")
 
-
     def on_job_done(self, job: BatchJob, row):
         """
         Handles jobs that have finished. Can be overridden to provide custom behaviour.

openeo/extra/job_management/_thread_worker.py

@@ -35,7 +35,7 @@ class _TaskResult:
     """
 
     job_id: str  # Mandatory
-    df_idx: int # Mandatory
+    df_idx: int  # Mandatory
     db_update: Dict[str, Any] = field(default_factory=dict)  # Optional
     stats_update: Dict[str, int] = field(default_factory=dict)  # Optional
 
@@ -118,7 +118,7 @@ def execute(self) -> _TaskResult:
             _log.info(f"Job {self.job_id!r} started successfully")
             return _TaskResult(
                 job_id=self.job_id,
-                df_idx = self.df_idx,
+                df_idx=self.df_idx,
                 db_update={"status": "queued"},
                 stats_update={"job start": 1},
             )
@@ -127,9 +127,9 @@ def execute(self) -> _TaskResult:
             # TODO: more insights about the failure (e.g. the exception) are just logged, but lost from the result
             return _TaskResult(
                 job_id=self.job_id,
-                df_idx = self.df_idx,
+                df_idx=self.df_idx,
                 db_update={"status": "start_failed"},
-                stats_update={"start_job error": 1}
+                stats_update={"start_job error": 1},
             )
 
 

tests/extra/job_management/test_job_management.py

@@ -94,15 +94,14 @@ class DummyTask(Task):
     """
 
     def __init__(self, job_id, df_idx, db_update, stats_update):
-        super().__init__(job_id=job_id, df_idx = df_idx)
+        super().__init__(job_id=job_id, df_idx=df_idx)
         self._db_update = db_update or {}
         self._stats_update = stats_update or {}
 
     def execute(self) -> _TaskResult:
-
         return _TaskResult(
             job_id=self.job_id,
-            df_idx = self.df_idx,
+            df_idx=self.df_idx,
             db_update=self._db_update,
             stats_update=self._stats_update,
         )
@@ -759,10 +758,12 @@ def test_process_threadworker_updates(self, tmp_path, caplog):
         # Invalid index (not in DB)
         pool.submit_task(DummyTask("j-missing", df_idx=4, db_update={"status": "created"}, stats_update=None))
 
-        df_initial = pd.DataFrame({
-            "id": ["j-0", "j-1", "j-2", "j-3"],
-            "status": ["created", "created", "created", "created"],
-        })
+        df_initial = pd.DataFrame(
+            {
+                "id": ["j-0", "j-1", "j-2", "j-3"],
+                "status": ["created", "created", "created", "created"],
+            }
+        )
         job_db = CsvJobDatabase(tmp_path / "jobs.csv").initialize_from_df(df_initial)
 
         mgr = MultiBackendJobManager(root_dir=tmp_path / "jobs")
@@ -786,7 +787,7 @@ def test_process_threadworker_updates(self, tmp_path, caplog):
         assert stats["job_db persist"] == 1
 
         # Assert error log for invalid index
-        assert any("Skipping non-existing dataframe indiches" in msg for msg in caplog.messages)
+        assert any("Skipping non-existing dataframe indices" in msg for msg in caplog.messages)
 
     def test_no_results_leaves_db_and_stats_untouched(self, tmp_path, caplog):
         pool = _JobManagerWorkerThreadPool(max_workers=2)
@@ -802,7 +803,6 @@ def test_no_results_leaves_db_and_stats_untouched(self, tmp_path, caplog):
         assert df_final.loc[0, "status"] == "created"
         assert stats == {}
 
-
     def test_logs_on_invalid_update(self, tmp_path, caplog):
         pool = _JobManagerWorkerThreadPool(max_workers=2)
         stats = collections.defaultdict(int)

tests/extra/job_management/test_thread_worker.py

@@ -25,9 +25,9 @@ def dummy_backend(requests_mock) -> DummyBackend:
 
 class TestTaskResult:
     def test_default(self):
-        result = _TaskResult(job_id="j-123", df_idx = 0)
+        result = _TaskResult(job_id="j-123", df_idx=0)
         assert result.job_id == "j-123"
-        assert result.df_idx ==0
+        assert result.df_idx == 0
         assert result.db_update == {}
         assert result.stats_update == {}
 
@@ -37,12 +37,14 @@ def test_start_success(self, dummy_backend, caplog):
         caplog.set_level(logging.WARNING)
         job = dummy_backend.connection.create_job(process_graph={})
 
-        task = _JobStartTask(job_id=job.job_id, df_idx=0, root_url=dummy_backend.connection.root_url, bearer_token="h4ll0")
+        task = _JobStartTask(
+            job_id=job.job_id, df_idx=0, root_url=dummy_backend.connection.root_url, bearer_token="h4ll0"
+        )
         result = task.execute()
 
         assert result == _TaskResult(
             job_id="job-000",
-            df_idx = 0,
+            df_idx=0,
             db_update={"status": "queued"},
             stats_update={"job start": 1},
         )
@@ -54,7 +56,9 @@ def test_start_failure(self, dummy_backend, caplog):
         job = dummy_backend.connection.create_job(process_graph={})
         dummy_backend.setup_job_start_failure()
 
-        task = _JobStartTask(job_id=job.job_id, df_idx=0, root_url=dummy_backend.connection.root_url, bearer_token="h4ll0")
+        task = _JobStartTask(
+            job_id=job.job_id, df_idx=0, root_url=dummy_backend.connection.root_url, bearer_token="h4ll0"
+        )
         result = task.execute()
 
         assert result == _TaskResult(
@@ -77,8 +81,6 @@ def test_hide_token(self, serializer):
         assert secret not in serialized
 
 
-
-
 class NopTask(Task):
     """Do Nothing"""
 
@@ -115,8 +117,6 @@ def execute(self) -> _TaskResult:
         return _TaskResult(job_id=self.job_id, df_idx=self.df_idx, db_update={"status": "all fine"})
 
 
-
-
 class TestJobManagerWorkerThreadPool:
     @pytest.fixture
     def worker_pool(self) -> Iterator[_JobManagerWorkerThreadPool]:
@@ -154,7 +154,7 @@ def test_submit_and_process_with_error(self, worker_pool):
         assert results == [
             _TaskResult(
                 job_id="j-666",
-                df_idx = 0,
+                df_idx=0,
                 db_update={"status": "threaded task failed"},
                 stats_update={"threaded task failed": 1},
             ),
@@ -171,7 +171,7 @@ def test_submit_and_process_iterative(self, worker_pool):
         worker_pool.submit_task(NopTask(job_id="j-22", df_idx=22))
         worker_pool.submit_task(NopTask(job_id="j-222", df_idx=222))
         results, remaining = worker_pool.process_futures(timeout=1)
-        assert results == [_TaskResult(job_id="j-22", df_idx=22), _TaskResult(job_id="j-222",  df_idx=222)]
+        assert results == [_TaskResult(job_id="j-22", df_idx=22), _TaskResult(job_id="j-222", df_idx=222)]
         assert remaining == 0
 
     def test_submit_multiple_simple(self, worker_pool):
@@ -212,7 +212,7 @@ def test_submit_multiple_blocking_and_failing(self, worker_pool):
         events[0].set()
         results, remaining = worker_pool.process_futures(timeout=0.1)
         assert results == [
-            _TaskResult(job_id="j-0", df_idx = 0, db_update={"status": "all fine"}),
+            _TaskResult(job_id="j-0", df_idx=0, db_update={"status": "all fine"}),
         ]
         assert remaining == n - 1
 
@@ -221,10 +221,13 @@ def test_submit_multiple_blocking_and_failing(self, worker_pool):
             events[j].set()
         results, remaining = worker_pool.process_futures(timeout=0.1)
         assert results == [
-            _TaskResult(job_id="j-1", df_idx = 1, db_update={"status": "all fine"}),
-            _TaskResult(job_id="j-2", df_idx = 2, db_update={"status": "all fine"}),
+            _TaskResult(job_id="j-1", df_idx=1, db_update={"status": "all fine"}),
+            _TaskResult(job_id="j-2", df_idx=2, db_update={"status": "all fine"}),
             _TaskResult(
-                job_id="j-3", df_idx = 3, db_update={"status": "threaded task failed"}, stats_update={"threaded task failed": 1}
+                job_id="j-3",
+                df_idx=3,
+                db_update={"status": "threaded task failed"},
+                stats_update={"threaded task failed": 1},
             ),
         ]
         assert remaining == 1
@@ -234,7 +237,7 @@ def test_submit_multiple_blocking_and_failing(self, worker_pool):
             events[j].set()
         results, remaining = worker_pool.process_futures(timeout=0.1)
         assert results == [
-            _TaskResult(job_id="j-4", df_idx = 4, db_update={"status": "all fine"}),
+            _TaskResult(job_id="j-4", df_idx=4, db_update={"status": "all fine"}),
         ]
         assert remaining == 0
 
@@ -260,7 +263,7 @@ def test_job_start_task(self, worker_pool, dummy_backend, caplog):
         assert results == [
             _TaskResult(
                 job_id="job-000",
-                df_idx = 0,
+                df_idx=0,
                 db_update={"status": "queued"},
                 stats_update={"job start": 1},
             )
@@ -278,7 +281,9 @@ def test_job_start_task_failure(self, worker_pool, dummy_backend, caplog):
 
         results, remaining = worker_pool.process_futures(timeout=1)
         assert results == [
-            _TaskResult(job_id="job-000", df_idx=0, db_update={"status": "start_failed"}, stats_update={"start_job error": 1})
+            _TaskResult(
+                job_id="job-000", df_idx=0, db_update={"status": "start_failed"}, stats_update={"start_job error": 1}
+            )
         ]
         assert remaining == 0
         assert caplog.messages == [