2025 housekeeping iv (#197)

* 2025_housekeeping_IV: duplicate writes

* 2025_housekeeping_IV: remove model writes

* Apply suggestions from code review

Author: DanielElisenberg

job_executor/adapter/local_storage/models/datastore_versions.py

@@ -1,7 +1,6 @@
 from datetime import UTC, datetime
 from typing import Iterator
 
-from job_executor.adapter import local_storage
 from job_executor.common.exceptions import (
     BumpException,
     ExistingDraftException,
@@ -117,7 +116,6 @@ def add(self, data_structure_update: DataStructureUpdate) -> None:
         self.data_structure_updates.append(data_structure_update)
         self._set_release_time()
         self._calculate_update_type()
-        self._write_to_file()
 
     def delete_draft(self, dataset_name: str) -> DataStructureUpdate:
         deleted_draft = next(
@@ -139,7 +137,6 @@ def delete_draft(self, dataset_name: str) -> DataStructureUpdate:
         ]
         self._set_release_time()
         self._calculate_update_type()
-        self._write_to_file()
         return deleted_draft
 
     def validate_bump_manifesto(
@@ -176,12 +173,8 @@ def release_pending(self) -> tuple[list[DataStructureUpdate], str]:
         self.data_structure_updates = draft_updates
         self._set_release_time()
         self._calculate_update_type()
-        self._write_to_file()
         return pending_updates, update_type
 
-    def _write_to_file(self) -> None:
-        local_storage.write_draft_version(self)
-
     def set_draft_release_status(
         self, dataset_name: str, new_status: str
     ) -> None:
@@ -199,7 +192,6 @@ def set_draft_release_status(
         dataset_update.set_release_status(new_status)
         self._set_release_time()
         self._calculate_update_type()
-        self._write_to_file()
 
     def _set_release_time(self) -> None:
         self.release_time = self._get_current_epoch_seconds()
@@ -214,9 +206,6 @@ class DatastoreVersions(CamelModel, extra="forbid"):
     description: str
     versions: list[DatastoreVersion]
 
-    def _write_to_file(self) -> None:
-        local_storage.write_datastore_versions(self)
-
     def _get_current_epoch_seconds(self) -> int:
         return int(
             (
@@ -260,7 +249,6 @@ def add_new_release_version(
             data_structure_updates=released_data_structure_updates,
         )
         self.versions = [new_release_version] + self.versions
-        self._write_to_file()
         return dotted_to_underscored_version(new_version_number)
 
     def get_dataset_release_status(self, dataset_name: str) -> str | None:
@@ -291,7 +279,12 @@ def bump_dotted_version_number(version: str, update_type: str) -> str:
         return ".".join([str(version_list[0] + 1), "0", "0", "0"])
     elif update_type == "MINOR":
         return ".".join(
-            [str(version_list[0]), str(version_list[1] + 1), "0", "0"]
+            [
+                str(version_list[0]),
+                str(version_list[1] + 1),
+                "0",
+                "0",
+            ]
         )
     elif update_type == "PATCH":
         return ".".join(

job_executor/adapter/local_storage/models/metadata.py

@@ -1,6 +1,5 @@
 from collections.abc import Iterator
 
-from job_executor.adapter import local_storage
 from job_executor.adapter.local_storage.models.datastore_versions import (
     DatastoreVersion,
 )
@@ -408,16 +407,12 @@ def get(self, dataset_name: str) -> Metadata | None:
 
 
 class MetadataAllDraft(MetadataAll):
-    def _write_to_file(self) -> None:
-        local_storage.write_metadata_all_draft(self)
-
     def remove(self, dataset_name: str) -> None:
         self.data_structures = [
             metadata
             for metadata in self.data_structures
             if metadata.name != dataset_name
         ]
-        self._write_to_file()
 
     def update_one(self, dataset_name: str, metadata: Metadata) -> None:
         self.data_structures = [
@@ -426,15 +421,12 @@ def update_one(self, dataset_name: str, metadata: Metadata) -> None:
             if metadata.name != dataset_name
         ]
         self.data_structures.append(metadata)
-        self._write_to_file()
 
     def remove_all(self) -> None:
         self.data_structures = []
-        self._write_to_file()
 
     def add(self, metadata: Metadata) -> None:
         self.data_structures.append(metadata)
-        self._write_to_file()
 
     def rebuild(
         self,
@@ -459,4 +451,3 @@ def rebuild(
                     )
                 new_data_structures[draft.name] = draft_metadata
         self.data_structures = list(new_data_structures.values())
-        self._write_to_file()

job_executor/domain/datastores.py

@@ -175,6 +175,7 @@ def patch_metadata(
             )
         patched_metadata = released_metadata.patch(draft_metadata)
         datastore.metadata_all_draft.update_one(dataset_name, patched_metadata)
+        local_storage.write_metadata_all_draft(datastore.metadata_all_draft)
         datastore.draft_version.add(
             DataStructureUpdate(
                 name=dataset_name,
@@ -183,6 +184,7 @@ def patch_metadata(
                 release_status="DRAFT",
             )
         )
+        local_storage.write_draft_version(datastore.draft_version)
         logger.info(f"{job_id}: completed")
         datastore_api.update_job_status(job_id, JobStatus.COMPLETED)
         logger.info(f"{job_id}: Deleting temporary backup")
@@ -230,9 +232,11 @@ def add(
                 release_status="DRAFT",
             )
         )
+        local_storage.write_draft_version(datastore.draft_version)
         draft_metadata = local_storage.get_working_dir_metadata(dataset_name)
         local_storage.make_dataset_dir(dataset_name)
         datastore.metadata_all_draft.add(draft_metadata)
+        local_storage.write_metadata_all_draft(datastore.metadata_all_draft)
         local_storage.move_working_dir_parquet_to_datastore(dataset_name)
         logger.info(f"{job_id}: completed")
         datastore_api.update_job_status(job_id, JobStatus.COMPLETED)
@@ -269,6 +273,7 @@ def change(
             )
         draft_metadata = local_storage.get_working_dir_metadata(dataset_name)
         datastore.metadata_all_draft.update_one(dataset_name, draft_metadata)
+        local_storage.write_metadata_all_draft(datastore.metadata_all_draft)
         datastore.draft_version.add(
             DataStructureUpdate(
                 name=dataset_name,
@@ -277,6 +282,7 @@ def change(
                 release_status="DRAFT",
             )
         )
+        local_storage.write_draft_version(datastore.draft_version)
         local_storage.move_working_dir_parquet_to_datastore(dataset_name)
         logger.info(f"{job_id}: completed")
         datastore_api.update_job_status(job_id, JobStatus.COMPLETED)
@@ -307,6 +313,7 @@ def remove(
     )
     if dataset_is_draft and dataset_operation == "REMOVE":
         datastore.metadata_all_draft.remove(dataset_name)
+        local_storage.write_metadata_all_draft(datastore.metadata_all_draft)
         log_message = "Dataset already in draft with operation REMOVE."
         logger.info(f"{job_id}: {log_message}")
         datastore_api.update_job_status(
@@ -321,6 +328,7 @@ def remove(
         datastore_api.update_job_status(job_id, JobStatus.FAILED, log_message)
     else:
         datastore.metadata_all_draft.remove(dataset_name)
+        local_storage.write_metadata_all_draft(datastore.metadata_all_draft)
         datastore.draft_version.add(
             DataStructureUpdate(
                 name=dataset_name,
@@ -329,6 +337,7 @@ def remove(
                 release_status="PENDING_DELETE",
             )
         )
+        local_storage.write_draft_version(datastore.draft_version)
         datastore_api.update_job_status(job_id, JobStatus.COMPLETED)
         logger.info(f"{job_id}: completed")
 
@@ -372,11 +381,14 @@ def delete_draft(
             raise VersioningException(log_message)
         datastore.metadata_all_draft.remove(dataset_name)
         datastore.metadata_all_draft.add(released_metadata)
+        local_storage.write_metadata_all_draft(datastore.metadata_all_draft)
     if dataset_operation == "ADD":
         datastore.metadata_all_draft.remove(dataset_name)
+        local_storage.write_metadata_all_draft(datastore.metadata_all_draft)
     if dataset_operation in ["ADD", "CHANGE"]:
         local_storage.delete_parquet_draft(dataset_name)
     datastore.draft_version.delete_draft(dataset_name)
+    local_storage.write_draft_version(datastore.draft_version)
     datastore_api.update_job_status(job_id, JobStatus.COMPLETED)
 
 
@@ -392,6 +404,7 @@ def set_draft_release_status(
         datastore.draft_version.set_draft_release_status(
             dataset_name, new_status
         )
+        local_storage.write_draft_version(datastore.draft_version)
         datastore_api.update_job_status(job_id, JobStatus.COMPLETED)
         logger.info(f"{job_id}: completed")
     except UnnecessaryUpdateException as e:
@@ -440,12 +453,14 @@ def bump_version(
 
         logger.info(f"{job_id}: Release pending operations from draft_version")
         release_updates, update_type = datastore.draft_version.release_pending()
+        local_storage.write_draft_version(datastore.draft_version)
         # If there are no released versions update type is MAJOR
         if datastore.metadata_all_latest is None:
             update_type = "MAJOR"
         new_version = datastore.datastore_versions.add_new_release_version(
             release_updates, description, update_type
         )
+        local_storage.write_datastore_versions(datastore.datastore_versions)
         logger.info(
             f"{job_id}: "
             f"Bumping from {datastore.latest_version_number} => {new_version}"
@@ -473,7 +488,7 @@ def bump_version(
             datastore.metadata_all_latest.data_structures,
             datastore.draft_version,
         )
-
+        local_storage.write_metadata_all_draft(datastore.metadata_all_draft)
         logger.info(f"{job_id}: completed BUMP")
         datastore_api.update_job_status(job_id, JobStatus.COMPLETED)
         logger.info(f"{job_id}: Archiving temporary backup")

tests/unit/model/test_datastore_version.py

@@ -103,6 +103,7 @@ def test_draft_version_delete_draft():
     release_time = draft_version.release_time
     version = draft_version.version
     draft_version.delete_draft("BRUTTO_INNTEKT")
+    local_storage.write_draft_version(draft_version)
     draft_version_file = load_json(DRAFT_VERSION_PATH)
     update_names = [
         update["name"] for update in draft_version_file["dataStructureUpdates"]