PT-5053 OGC Job Active Record (#623)

Adding Job active record to the processing module

CHANGELOG.md

@@ -29,6 +29,12 @@ You can check your current version with the following command:
     ```
 
 For more information, see [UP42 Python package description](https://pypi.org/project/up42-py/).
+# 1.0.4a14
+
+**Jun 10, 2024**
+
+- Added class `Job` to `processing` module to access processing job features.
+
 ## 1.0.4a13
 
 **June 10, 2024**

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "up42-py"
-version = "1.0.4a13"
+version = "1.0.4a14"
 description = "Python SDK for UP42, the geospatial marketplace and developer platform."
 authors = ["UP42 GmbH <[email protected]>"]
 license = "https://github.com/up42/up42-py/blob/master/LICENSE"

tests/test_processing.py

@@ -1,5 +1,7 @@
 import dataclasses
+import datetime
 import random
+import uuid
 from unittest import mock
 
 import pystac
@@ -30,6 +32,41 @@
     }
 )
 
+JOB_ID = str(uuid.uuid4())
+GET_JOB_URL = f"{constants.API_HOST}/v2/processing/jobs/{JOB_ID}"
+CREDITS = 1
+ACCOUNT_ID = str(uuid.uuid4())
+DEFINITION = {
+    "inputs": {
+        "item": ITEM_URL,
+        "title": TITLE,
+    }
+}
+NOW = datetime.datetime.now()
+JOB_METADATA: processing.JobMetadata = {
+    "processID": PROCESS_ID,
+    "jobID": JOB_ID,
+    "accountID": ACCOUNT_ID,
+    "workspaceID": constants.WORKSPACE_ID,
+    "definition": DEFINITION,
+    "status": "created",
+    "created": f"{NOW.isoformat()}Z",
+    "updated": f"{NOW.isoformat()}Z",
+    "started": None,
+    "finished": None,
+}
+
+JOB = processing.Job(
+    process_id=PROCESS_ID,
+    id=JOB_ID,
+    account_id=ACCOUNT_ID,
+    workspace_id=constants.WORKSPACE_ID,
+    definition=DEFINITION,
+    status=processing.JobStatus.CREATED,
+    created=NOW,
+    updated=NOW,
+)
+
 
 @pytest.fixture(autouse=True)
 def workspace():
@@ -213,3 +250,9 @@ def test_should_provide_inputs(self, requests_mock: req_mock.Mocker):
         assert template.is_valid
         assert template.cost == cost
         assert template.inputs == {"title": TITLE, "items": [ITEM_URL]}
+
+
+class TestJob:
+    def test_should_get_job(self, requests_mock: req_mock.Mocker):
+        requests_mock.get(url=GET_JOB_URL, json=JOB_METADATA)
+        assert processing.Job.get(JOB_ID) == JOB

up42/processing.py

@@ -1,6 +1,8 @@
 import abc
 import dataclasses
-from typing import ClassVar, List, Optional, Union
+import datetime
+import enum
+from typing import ClassVar, List, Optional, TypedDict, Union
 
 import pystac
 import requests
@@ -116,3 +118,69 @@ def inputs(self) -> dict:
             "title": self.title,
             "items": [item.get_self_href() for item in self.items],
         }
+
+
+class JobStatus(enum.Enum):
+    CREATED = "created"
+    VALID = "valid"
+    INVALID = "invalid"
+    ACCEPTED = "accepted"
+    REJECTED = "rejected"
+    RUNNING = "running"
+    SUCCESSFUL = "successful"
+    FAILED = "failed"
+    CAPTURED = "captured"
+    RELEASED = "released"
+
+
+class JobMetadata(TypedDict):
+    processID: str  # pylint: disable=invalid-name
+    jobID: str  # pylint: disable=invalid-name
+    accountID: str  # pylint: disable=invalid-name
+    workspaceID: Optional[str]  # pylint: disable=invalid-name
+    definition: dict
+    status: str
+    created: str
+    started: Optional[str]
+    finished: Optional[str]
+    updated: str
+
+
+@dataclasses.dataclass
+class Job:
+    session = base.Session()
+    process_id: str
+    id: str
+    account_id: str
+    workspace_id: Optional[str]
+    definition: dict
+    status: JobStatus
+    created: datetime.datetime
+    updated: datetime.datetime
+    started: Optional[datetime.datetime] = None
+    finished: Optional[datetime.datetime] = None
+
+    @staticmethod
+    def __to_datetime(value: Optional[str]):
+        return value and datetime.datetime.fromisoformat(value.rstrip("Z"))
+
+    @staticmethod
+    def from_metadata(metadata: JobMetadata) -> "Job":
+        return Job(
+            process_id=metadata["processID"],
+            id=metadata["jobID"],
+            account_id=metadata["accountID"],
+            workspace_id=metadata["workspaceID"],
+            definition=metadata["definition"],
+            status=JobStatus(metadata["status"]),
+            created=Job.__to_datetime(metadata["created"]),
+            started=Job.__to_datetime(metadata["started"]),
+            finished=Job.__to_datetime(metadata["finished"]),
+            updated=Job.__to_datetime(metadata["updated"]),
+        )
+
+    @classmethod
+    def get(cls, job_id: str) -> "Job":
+        url = host.endpoint(f"/v2/processing/jobs/{job_id}")
+        metadata = cls.session.get(url).json()
+        return cls.from_metadata(metadata)