[TS-1719] Feasibility active record class (#750)

Author: gustavo-hsu-wu

CHANGELOG.md

@@ -30,6 +30,11 @@ You can check your current version with the following command:
 
 For more information, see [UP42 Python package description](https://pypi.org/project/up42-py/).
 
+### 2.4.0a1
+**July 02, 2025**
+- Added `FeasibilityStudy` active record class to `tasking` module.
+- Exported `FeasibilityStudy` and `FeasibilityStudySorting` to `up42` namespace.
+
 ## 2.3.1
 **June 26, 2025**
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "up42-py"
-version = "2.3.1"
+version = "2.4.0a1"
 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_tasking.py

@@ -345,3 +345,118 @@ def test_should_get_all(
             sort_by=sort_by,
         )
         assert list(quotations) == [self.quotation] * 4
+
+
+class TestFeasibilityStudy:
+    FEASIBILITY_ID: str = FEASIBILITY_ID
+    ACCOUNT_ID: str = ACCOUNT_ID
+    WORKSPACE_ID: str = "workspace-id"
+    ORDER_ID: str = "test-order-id"
+    OPTION_ID: str = "option-123"
+
+    metadata: dict = {
+        "id": FEASIBILITY_ID,
+        "createdAt": "created-at",
+        "updatedAt": "updated-at",
+        "accountId": ACCOUNT_ID,
+        "workspaceId": WORKSPACE_ID,
+        "orderId": ORDER_ID,
+        "decision": "NOT_DECIDED",
+        "options": [{"id": OPTION_ID}],
+        "decisionAt": "decided-at",
+    }
+    feasibility_study = tasking.FeasibilityStudy(
+        id=FEASIBILITY_ID,
+        created_at="created-at",
+        updated_at="updated-at",
+        account_id=ACCOUNT_ID,
+        workspace_id=WORKSPACE_ID,
+        order_id=ORDER_ID,
+        decision="NOT_DECIDED",
+        options=[{"id": OPTION_ID}],
+        decided_at="decided-at",
+        decision_option=None,
+    )
+
+    @pytest.mark.parametrize("feasibility_study_id", [None, FEASIBILITY_ID])
+    @pytest.mark.parametrize("workspace_id", [None, WORKSPACE_ID])
+    @pytest.mark.parametrize("order_id", [None, ORDER_ID])
+    @pytest.mark.parametrize("decision", [None, ["NOT_DECIDED", "ACCEPTED"], ["ACCEPTED"]])
+    @pytest.mark.parametrize(
+        "sort_by",
+        [
+            None,
+            tasking.FeasibilityStudySorting.created_at.asc,
+            tasking.FeasibilityStudySorting.updated_at.asc,
+            tasking.FeasibilityStudySorting.decided_at.asc,
+        ],
+        ids=str,
+    )
+    def test_should_get_all(
+        self,
+        requests_mock: req_mock.Mocker,
+        feasibility_study_id: Optional[str],
+        workspace_id: Optional[str],
+        order_id: Optional[str],
+        decision: Optional[List[tasking.FeasibilityStatus]],
+        sort_by: Optional[utils.SortingField],
+    ):
+        query_params: dict[str, Any] = {}
+        if feasibility_study_id:
+            query_params["id"] = feasibility_study_id
+        if workspace_id:
+            query_params["workspaceId"] = workspace_id
+        if order_id:
+            query_params["orderId"] = order_id
+        if decision:
+            query_params["decision"] = decision
+        if sort_by:
+            query_params["sort"] = str(sort_by)
+        base_url = f"{constants.API_HOST}/v2/tasking/feasibility-studies"
+        expected = [self.metadata] * 4
+        for page in [0, 1]:
+            query_params["page"] = page
+            query = urllib.parse.urlencode(query_params, doseq=True, safe="")
+            url = base_url + (query and f"?{query}")
+            offset = page * 2
+            response = {
+                "content": expected[offset : offset + 2],  # noqa: E203
+                "totalPages": 2,
+            }
+            requests_mock.get(url=url, json=response)
+        feasibility_studies = tasking.FeasibilityStudy.all(
+            feasibility_study_id=feasibility_study_id,
+            workspace_id=workspace_id,
+            order_id=order_id,
+            decision=decision,
+            sort_by=sort_by,
+        )
+        assert list(feasibility_studies) == [self.feasibility_study] * 4
+
+    def test_should_accept(self):
+        feasibility_study = dataclasses.replace(self.feasibility_study, decision_option=None)
+        feasibility_study.accept(self.OPTION_ID)
+        assert feasibility_study.decision_option.id == self.OPTION_ID  # type: ignore[union-attr]
+
+    def test_should_save(self, requests_mock: req_mock.Mocker):
+        feasibility_study = dataclasses.replace(self.feasibility_study, decision_option=None)
+        feasibility_study.accept(self.OPTION_ID)
+        patch = {"acceptedOptionId": self.OPTION_ID}
+        url = f"{constants.API_HOST}/v2/tasking/feasibility-studies/{self.FEASIBILITY_ID}"
+        expected_description = "description"
+        expected_json = self.metadata | {"decisionOption": {"id": self.OPTION_ID, "description": expected_description}}
+        requests_mock.patch(
+            url=url,
+            json=expected_json,
+            additional_matcher=helpers.match_request_body(patch),
+        )
+        feasibility_study.save()
+        assert feasibility_study.decision_option.id == self.OPTION_ID  # type: ignore[union-attr]
+        assert feasibility_study.decision_option.description == expected_description  # type: ignore[union-attr]
+
+    def test_should_raise_if_no_decision_option_on_save(self):
+        feasibility_study = dataclasses.replace(self.feasibility_study, decision_option=None)
+        with pytest.raises(
+            feasibility_study.NoDecisionOptionChosen, match="No decision option chosen for this feasibility study."
+        ):
+            feasibility_study.save()

up42/__init__.py

@@ -36,7 +36,7 @@
 from up42.processing import Job, JobSorting, JobStatus
 from up42.stac import extend as stac_extend
 from up42.storage import Storage
-from up42.tasking import Quotation, QuotationSorting, Tasking
+from up42.tasking import FeasibilityStudy, FeasibilityStudySorting, Quotation, QuotationSorting, Tasking
 from up42.tools import get_example_aoi, read_vector_file
 from up42.utils import get_up42_py_version
 from up42.webhooks import Webhook
@@ -77,5 +77,7 @@
         BatchOrderTemplate,
         Quotation,
         QuotationSorting,
+        FeasibilityStudy,
+        FeasibilityStudySorting,
     ]
 ]

up42/tasking.py

@@ -155,7 +155,7 @@ def decide_quotation(self, quotation_id: str, decision: QuotationDecision) -> di
         url = host.endpoint(f"/v2/tasking/quotation/{quotation_id}")
         return self.session.patch(url, json={"decision": decision}).json()
 
-    @utils.deprecation(None, "3.0.0")
+    @utils.deprecation("FeasibilityStudy::all", "3.0.0")
     def get_feasibility(
         self,
         feasibility_id: Optional[str] = None,
@@ -190,7 +190,7 @@ def get_feasibility(
         }
         return list(utils.paged_query(params, "/v2/tasking/feasibility-studies", self.session))
 
-    @utils.deprecation(None, "3.0.0")
+    @utils.deprecation("FeasibilityStudy", "3.0.0")
     def choose_feasibility(self, feasibility_id: str, accepted_option_id: str) -> dict:
         """Accept one of the proposed feasibility study options.
         This operation is only allowed on feasibility studies with the NOT_DECIDED status.
@@ -273,3 +273,87 @@ def all(
             cls._from_metadata,
             utils.paged_query(params, "/v2/tasking/quotation", cls.session),
         )
+
+
+class FeasibilityStudySorting:
+    created_at = utils.SortingField(name="createdAt")
+    updated_at = utils.SortingField(name="updatedAt")
+    decided_at = utils.SortingField(name="decisionAt")
+
+
+@dataclasses.dataclass
+class FeasibilityStudyDecisionOption:
+    id: str
+    description: Optional[str] = None
+
+
+@dataclasses.dataclass
+class FeasibilityStudy:
+    session = base.Session()
+    id: str
+    created_at: str
+    updated_at: str
+    account_id: str
+    workspace_id: str
+    order_id: str
+    decision: FeasibilityStatus
+    options: List[dict]
+    decided_at: Optional[str] = None
+    decision_option: Optional[FeasibilityStudyDecisionOption] = None
+
+    class NoDecisionOptionChosen(Exception):
+        """Raised when trying to save a FeasibilityStudy without a chosen decision option."""
+
+    @classmethod
+    def all(
+        cls,
+        feasibility_study_id: Optional[str] = None,
+        workspace_id: Optional[str] = None,
+        order_id: Optional[str] = None,
+        decision: Optional[List[FeasibilityStatus]] = None,
+        sort_by: Optional[utils.SortingField] = None,
+    ) -> Iterator["FeasibilityStudy"]:
+        params = {
+            "id": feasibility_study_id,
+            "workspaceId": workspace_id,
+            "orderId": order_id,
+            "decision": decision,
+            "sort": sort_by,
+        }
+        return map(
+            cls._from_metadata,
+            utils.paged_query(params, "/v2/tasking/feasibility-studies", cls.session),
+        )
+
+    @staticmethod
+    def _from_metadata(metadata: dict) -> "FeasibilityStudy":
+        decision_option = metadata.get("decisionOption")
+        if decision_option is not None:
+            decision_option = FeasibilityStudyDecisionOption(decision_option["id"], decision_option["description"])
+        return FeasibilityStudy(
+            id=metadata["id"],
+            created_at=metadata["createdAt"],
+            updated_at=metadata["updatedAt"],
+            account_id=metadata["accountId"],
+            workspace_id=metadata["workspaceId"],
+            order_id=metadata["orderId"],
+            decision=metadata["decision"],
+            options=metadata.get("options", []),
+            decided_at=metadata.get("decisionAt"),
+            decision_option=decision_option,
+        )
+
+    def accept(self, option_id: str):
+        self.decision_option = FeasibilityStudyDecisionOption(option_id)
+
+    def save(self):
+        url = host.endpoint(f"/v2/tasking/feasibility-studies/{self.id}")
+        if self.decision_option is None:
+            raise FeasibilityStudy.NoDecisionOptionChosen(
+                "No decision option chosen for this feasibility study. "
+                "Please call 'accept' with a valid option ID before saving."
+            )
+        metadata = self.session.patch(url, json={"acceptedOptionId": self.decision_option.id}).json()
+        feasibility_study = self._from_metadata(metadata)
+        for field in dataclasses.fields(feasibility_study):
+            setattr(self, field.name, getattr(feasibility_study, field.name))