Remove cached property from JobProgressEncoder (#3442)

## Changes
Remove cached property from JobProgressEncoder

### Linked issues

Resolves #3377

Author: JCZuurmond

src/databricks/labs/ucx/progress/jobs.py

@@ -1,16 +1,22 @@
 import collections
+import logging
+from collections.abc import Iterable
 from dataclasses import replace
-from functools import cached_property
 
 from databricks.labs.lsql.backends import SqlBackend
 
 from databricks.labs.ucx.assessment.jobs import JobInfo, JobOwnership
+from databricks.labs.ucx.framework.utils import escape_sql_identifier
 from databricks.labs.ucx.progress.history import ProgressEncoder
 from databricks.labs.ucx.progress.install import Historical
 from databricks.labs.ucx.source_code.directfs_access import DirectFsAccessCrawler
 from databricks.labs.ucx.source_code.jobs import JobProblem
 
 
+logger = logging.getLogger(__name__)
+JobIdToFailuresType = dict[str, list[str]]  # dict[<job id>, list[<failure message>]]
+
+
 class JobsProgressEncoder(ProgressEncoder[JobInfo]):
     """Encoder class:Job to class:History."""
 
@@ -37,8 +43,18 @@ def __init__(
         self._direct_fs_access_crawlers = direct_fs_access_crawlers
         self._inventory_database = inventory_database
 
-    @cached_property
-    def _job_problems(self) -> dict[int, list[str]]:
+    def append_inventory_snapshot(self, snapshot: Iterable[JobInfo]) -> None:
+        job_problems = self._get_job_problems()
+        dfsas = self._get_direct_filesystem_accesses()
+        history_records = []
+        for record in snapshot:
+            history_record = self._encode_job_as_historical(record, job_problems, dfsas)
+            history_records.append(history_record)
+        logger.debug(f"Appending {len(history_records)} {self._klass} table record(s) to history.")
+        # The mode is 'append'. This is documented as conflict-free.
+        self._sql_backend.save_table(escape_sql_identifier(self.full_name), history_records, Historical, mode="append")
+
+    def _get_job_problems(self) -> JobIdToFailuresType:
         index = collections.defaultdict(list)
         for row in self._sql_backend.fetch(
             'SELECT * FROM workflow_problems',
@@ -47,11 +63,10 @@ def _job_problems(self) -> dict[int, list[str]]:
         ):
             job_problem = JobProblem(**row.asDict())
             failure = f'{job_problem.code}: {job_problem.task_key} task: {job_problem.path}: {job_problem.message}'
-            index[job_problem.job_id].append(failure)
+            index[str(job_problem.job_id)].append(failure)
         return index
 
-    @cached_property
-    def _direct_fs_accesses(self) -> dict[str, list[str]]:
+    def _get_direct_filesystem_accesses(self) -> JobIdToFailuresType:
         index = collections.defaultdict(list)
         for crawler in self._direct_fs_access_crawlers:
             for direct_fs_access in crawler.snapshot():
@@ -71,7 +86,12 @@ def _direct_fs_accesses(self) -> dict[str, list[str]]:
                 index[job_id].append(failure)
         return index
 
-    def _encode_record_as_historical(self, record: JobInfo) -> Historical:
+    def _encode_job_as_historical(
+        self,
+        record: JobInfo,
+        job_problems: JobIdToFailuresType,
+        dfsas: JobIdToFailuresType,
+    ) -> Historical:
         """Encode a job as a historical records.
 
         Failures are detected by the WorkflowLinter:
@@ -80,6 +100,6 @@ def _encode_record_as_historical(self, record: JobInfo) -> Historical:
         """
         historical = super()._encode_record_as_historical(record)
         failures = []
-        failures.extend(self._job_problems.get(int(record.job_id), []))
-        failures.extend(self._direct_fs_accesses.get(record.job_id, []))
+        failures.extend(job_problems.get(record.job_id, []))
+        failures.extend(dfsas.get(record.job_id, []))
         return replace(historical, failures=historical.failures + failures)