200 new job model (#206)

* 200-new-job-model: update config

* 200-new-job-model: update datastore-api adapter

* 200-new-job-model: test adapter

* 200-new-job-model: domain tests

* 200-new-job-model: datastore specific localstorage adapters

Author: DanielElisenberg

.test.env

@@ -1,4 +1,5 @@
 DATASTORE_DIR=tests/resources/datastores/TEST_DATASTORE
+DATASTORE_RDN=no.ssb.test
 PSEUDONYM_SERVICE_URL=http://mock.pseudonym.service
 DATASTORE_API_URL=http://mock.job.service
 NUMBER_OF_WORKERS=4

job_executor/adapter/datastore_api/__init__.py

@@ -1,4 +1,5 @@
 import logging
+from pathlib import Path
 from urllib.error import HTTPError
 
 import requests
@@ -22,6 +23,37 @@
 logger = logging.getLogger()
 
 
+def execute_request(
+    method: str,
+    url: str,
+    retry: bool = False,
+    **kwargs,  # noqa
+) -> Response:
+    try:
+        if retry:
+            with requests.Session() as s:
+                retries = Retry(
+                    total=6,
+                    backoff_factor=0.5,
+                    # [0.0s, 1.0s, 2.0s, 4.0s, 8.0s, 16.0s] between retries
+                    allowed_methods={"GET"},
+                )
+                s.mount("http://", HTTPAdapter(max_retries=retries))
+                response = s.request(method=method, url=url, **kwargs)
+        else:
+            response = requests.request(
+                method=method,
+                url=url,
+                timeout=DEFAULT_REQUESTS_TIMEOUT,
+                **kwargs,
+            )
+        if response.status_code != 200:
+            raise HttpResponseError(f"{response.status_code}: {response.text}")
+        return response
+    except (RequestException, HTTPError) as e:
+        raise HttpRequestError(e) from e
+
+
 def get_jobs(
     job_status: JobStatus | None = None,
     operations: list[Operation] | None = None,
@@ -40,7 +72,10 @@ def get_jobs(
         request_url += f"?{'&'.join(query_fields)}"
 
     response = execute_request("GET", request_url, True)
-    return [Job(**job) for job in response.json()]
+    return [
+        Job.model_validate({"datastoreRdn": environment.datastore_rdn, **job})
+        for job in response.json()
+    ]
 
 
 def update_job_status(
@@ -72,35 +107,8 @@ def is_system_paused() -> bool:
     return maintenance_status.paused
 
 
-def execute_request(
-    method: str,
-    url: str,
-    retry: bool = False,
-    **kwargs,  # noqa
-) -> Response:
-    try:
-        if retry:
-            with requests.Session() as s:
-                retries = Retry(
-                    total=6,
-                    backoff_factor=0.5,
-                    # [0.0s, 1.0s, 2.0s, 4.0s, 8.0s, 16.0s] between retries
-                    allowed_methods={"GET"},
-                )
-                s.mount("http://", HTTPAdapter(max_retries=retries))
-                response = s.request(method=method, url=url, **kwargs)
-        else:
-            response = requests.request(
-                method=method,
-                url=url,
-                timeout=DEFAULT_REQUESTS_TIMEOUT,
-                **kwargs,
-            )
-        if response.status_code != 200:
-            raise HttpResponseError(f"{response.status_code}: {response.text}")
-        return response
-    except (RequestException, HTTPError) as e:
-        raise HttpRequestError(e) from e
+def get_datastore_directory(rdn: str) -> Path:
+    return Path(environment.datastore_dir)
 
 
 def query_for_jobs() -> JobQueryResult:

job_executor/adapter/datastore_api/models.py

@@ -90,6 +90,7 @@ class Log(CamelModel, extra="forbid"):
 
 class Job(CamelModel, use_enum_values=True):
     job_id: str
+    datastore_rdn: str
     status: JobStatus
     parameters: JobParameters
     log: list[Log] | None = []

job_executor/app.py

@@ -29,7 +29,6 @@ def initialize_app() -> None:
 def handle_jobs(manager: Manager, logging_queue: Queue) -> None:
     job_query_result = datastore_api.query_for_jobs()
     manager.clean_up_after_dead_workers()
-    local_storage = LocalStorageAdapter(Path(environment.datastore_dir))
     if job_query_result.available_jobs_count:
         logger.info(
             f"Found {len(job_query_result.queued_worker_jobs)}"
@@ -39,6 +38,9 @@ def handle_jobs(manager: Manager, logging_queue: Queue) -> None:
         )
 
     for job in job_query_result.queued_worker_jobs:
+        local_storage = LocalStorageAdapter(
+            datastore_api.get_datastore_directory(job.datastore_rdn)
+        )
         job_size = local_storage.input_dir.get_importable_tar_size_in_bytes(
             job.parameters.target
         )

job_executor/config/__init__.py

@@ -6,6 +6,7 @@
 @dataclass
 class Environment:
     datastore_dir: str
+    datastore_rdn: str
     secrets_file: str
     pseudonym_service_url: str
     datastore_api_url: str
@@ -18,6 +19,7 @@ class Environment:
 def _initialize_environment() -> Environment:
     return Environment(
         datastore_dir=os.environ["DATASTORE_DIR"],
+        datastore_rdn=os.environ["DATASTORE_RDN"],
         secrets_file=os.environ["SECRETS_FILE"],
         pseudonym_service_url=os.environ["PSEUDONYM_SERVICE_URL"],
         datastore_api_url=os.environ["DATASTORE_API_URL"],