187 do not persist datastore models in memory (#193)

* 187: read datastore files before each job

* 187: no need to refresh after job

* 187: remove filereading from models

* 187: linting

Author: DanielElisenberg

job_executor/app.py

@@ -6,7 +6,6 @@
 from job_executor.config import environment
 from job_executor.config.log import initialize_logging_thread, setup_logging
 from job_executor.domain import rollback
-from job_executor.domain.datastore import Datastore
 from job_executor.exception import StartupException
 from job_executor.manager import Manager
 from job_executor.model.job import JobStatus
@@ -86,7 +85,6 @@ def main() -> None:
                 int(environment.get("MAX_GB_ALL_WORKERS"))
                 * 1024**3  # Covert from GB to bytes
             ),
-            this_datastore=Datastore(),
         )
 
         while True:

job_executor/domain/datastore.py job_executor/domain/datastores.pyjob_executor/domain/datastore.py renamed to job_executor/domain/datastores.py

@@ -29,20 +29,23 @@ class Datastore:
     latest_version_number: str | None
 
     def __init__(self) -> None:
-        self.refresh()
-
-    def refresh(self) -> None:
-        self.draft_version = DraftVersion()  # type: ignore
-        self.datastore_versions = DatastoreVersions()  # type: ignore
+        self.draft_version = DraftVersion.model_validate(
+            local_storage.get_draft_version()
+        )
+        self.datastore_versions = DatastoreVersions.model_validate(
+            local_storage.get_datastore_versions()
+        )
         self.latest_version_number = (
             self.datastore_versions.get_latest_version_number()
         )
-        self.metadata_all_draft = MetadataAllDraft()  # type: ignore
+        self.metadata_all_draft = MetadataAllDraft.model_validate(
+            local_storage.get_metadata_all("DRAFT")
+        )
         if self.latest_version_number is None:
             self.metadata_all_latest = None
         else:
-            self.metadata_all_latest = MetadataAll(
-                **local_storage.get_metadata_all(self.latest_version_number)
+            self.metadata_all_latest = MetadataAll.model_validate(
+                local_storage.get_metadata_all(self.latest_version_number)
             )
 
 
@@ -197,15 +200,13 @@ def patch_metadata(
             job_id, "PATCH_METADATA", dataset_name
         )
         job_service.update_job_status(job_id, JobStatus.FAILED, str(e))
-        datastore.refresh()
     except Exception as e:
         logger.error(f"{job_id}: An unexpected error occured")
         logger.exception(f"{job_id}: {str(e)}", exc_info=e)
         rollback_manager_phase_import_job(
             job_id, "PATCH_METADATA", dataset_name
         )
         job_service.update_job_status(job_id, JobStatus.FAILED)
-        datastore.refresh()
 
 
 def add(
@@ -250,7 +251,6 @@ def add(
         logger.exception(f"{job_id}: {str(e)}", exc_info=e)
         rollback_manager_phase_import_job(job_id, "ADD", dataset_name)
         job_service.update_job_status(job_id, JobStatus.FAILED)
-        datastore.refresh()
 
 
 def change(
@@ -297,7 +297,6 @@ def change(
         logger.exception(f"{job_id}: {str(e)}", exc_info=e)
         rollback_manager_phase_import_job(job_id, "CHANGE", dataset_name)
         job_service.update_job_status(job_id, JobStatus.FAILED)
-        datastore.refresh()
 
 
 def remove(
@@ -491,7 +490,6 @@ def bump_version(
             bump_manifesto.model_dump(by_alias=True, exclude_none=True),
         )
         job_service.update_job_status(job_id, JobStatus.FAILED)
-        datastore.refresh()
 
 
 def delete_archived_input(job_id: str, dataset_name: str) -> None:

job_executor/manager/__init__.py

@@ -2,8 +2,8 @@
 from multiprocessing import Process, Queue
 
 from job_executor.adapter import job_service
-from job_executor.domain import datastore, rollback
-from job_executor.domain.datastore import Datastore
+from job_executor.domain import datastores, rollback
+from job_executor.domain.datastores import Datastore
 from job_executor.model.job import Job, JobStatus, Operation
 from job_executor.model.worker import Worker
 from job_executor.worker import build_dataset_worker, build_metadata_worker
@@ -26,7 +26,6 @@ def __init__(
         self,
         max_workers: int,
         max_bytes_all_workers: int,
-        this_datastore: Datastore,
     ) -> None:
         """
         :param default_max_workers: The maximum number of workers
@@ -37,7 +36,6 @@ def __init__(
         """
         self.max_workers = max_workers
         self.max_bytes_all_workers = max_bytes_all_workers
-        self.datastore = this_datastore
         self.workers: list[Worker] = []
 
     @property
@@ -141,69 +139,70 @@ def handle_worker_job(
     def handle_manager_job(self, job: Job) -> None:
         job_id = job.job_id
         operation = job.parameters.operation
+        datastore = Datastore()
         self.unregister_worker(
             job_id
         )  # Filter out job from worker jobs if built
         # Ignoring a lot of types here as we already have done the validation
         # in the pydantic model.
         if operation == Operation.BUMP:
-            datastore.bump_version(
-                self.datastore,
+            datastores.bump_version(
+                datastore,
                 job_id,
                 job.parameters.bump_manifesto,  # type: ignore
                 job.parameters.description,  # type: ignore
             )
         elif operation == Operation.PATCH_METADATA:
-            datastore.patch_metadata(
-                self.datastore,
+            datastores.patch_metadata(
+                datastore,
                 job_id,
                 job.parameters.target,
                 job.parameters.description,  # type: ignore
             )
         elif operation == Operation.SET_STATUS:
-            datastore.set_draft_release_status(
-                self.datastore,
+            datastores.set_draft_release_status(
+                datastore,
                 job_id,
                 job.parameters.target,
                 job.parameters.release_status,  # type: ignore
             )
         elif operation == Operation.ADD:
-            datastore.add(
-                self.datastore,
+            datastores.add(
+                datastore,
                 job_id,
                 job.parameters.target,
                 job.parameters.description,  # type: ignore
             )
         elif operation == Operation.CHANGE:
-            datastore.change(
-                self.datastore,
+            datastores.change(
+                datastore,
                 job_id,
                 job.parameters.target,
                 job.parameters.description,  # type: ignore
             )
         elif operation == Operation.REMOVE:
-            datastore.remove(
-                self.datastore,
+            datastores.remove(
+                datastore,
                 job_id,
                 job.parameters.target,
                 job.parameters.description,  # type: ignore
             )
         elif operation == Operation.ROLLBACK_REMOVE:
-            datastore.delete_draft(
-                self.datastore,
+            datastores.delete_draft(
+                datastore,
                 job_id,
                 job.parameters.target,
                 rollback_remove=True,
             )
         elif operation == Operation.DELETE_DRAFT:
-            datastore.delete_draft(
-                self.datastore,
+            datastores.delete_draft(
+                datastore,
                 job_id,
                 job.parameters.target,
                 rollback_remove=False,
             )
         elif operation == Operation.DELETE_ARCHIVE:
-            datastore.delete_archived_input(job_id, job.parameters.target)
+            datastores.delete_archived_input(job_id, job.parameters.target)
         else:
             job_service.update_job_status(
                 job.job_id, JobStatus.FAILED, log="Unknown operation for job"

job_executor/model/datastore_version.py

@@ -1,8 +1,6 @@
 from datetime import UTC, datetime
 from typing import Iterator
 
-from pydantic import model_validator
-
 from job_executor.adapter import local_storage
 from job_executor.exception import (
     BumpException,
@@ -77,11 +75,6 @@ def contains(self, dataset_name: str) -> bool:
 
 
 class DraftVersion(DatastoreVersion):
-    @model_validator(mode="before")
-    @classmethod
-    def read_file(cls, _):  # noqa
-        return local_storage.get_draft_version()
-
     def add(self, data_structure_update: DataStructureUpdate) -> None:
         current_update_names = [update.name for update in self]
         if data_structure_update.name in current_update_names:

job_executor/model/datastore_versions.py

@@ -1,7 +1,5 @@
 from datetime import UTC, datetime
 
-from pydantic import model_validator
-
 from job_executor.adapter import local_storage
 from job_executor.exception import VersioningException
 from job_executor.model.camelcase_model import CamelModel
@@ -15,11 +13,6 @@ class DatastoreVersions(CamelModel, extra="forbid"):
     description: str
     versions: list[DatastoreVersion]
 
-    @model_validator(mode="before")
-    @classmethod
-    def read_file(cls, _):  # noqa
-        return local_storage.get_datastore_versions()
-
     def _write_to_file(self) -> None:
         local_storage.write_datastore_versions(self.model_dump(by_alias=True))
 

job_executor/model/metadata_all.py

@@ -1,7 +1,5 @@
 from collections.abc import Iterator
 
-from pydantic import model_validator
-
 from job_executor.adapter import local_storage
 from job_executor.exception import BumpException
 from job_executor.model import Metadata
@@ -48,11 +46,6 @@ def get(self, dataset_name: str) -> Metadata | None:
 
 
 class MetadataAllDraft(MetadataAll):
-    @model_validator(mode="before")
-    @classmethod
-    def read_file(cls, _) -> dict:  # noqa
-        return local_storage.get_metadata_all("DRAFT")
-
     def _write_to_file(self) -> None:
         local_storage.write_metadata_all(
             self.model_dump(by_alias=True, exclude_none=True), "DRAFT"