Merge pull request #17 from Sage-Bionetworks-Workflows/bgrande/workflows-528/tweaks

[WORKFLOWS-528] Allow `LaunchInfo` values to be missing

Author: Bruno Grande

src/orca/services/nextflowtower/models.py

@@ -1,14 +1,13 @@
 import json
-from collections.abc import Collection
 from dataclasses import field
 from datetime import datetime
 from enum import Enum
-from typing import Any, Optional
+from typing import Any, Iterable, Optional
 
 from pydantic.dataclasses import dataclass
 from typing_extensions import Self
 
-from orca.services.nextflowtower.utils import parse_datetime
+from orca.services.nextflowtower.utils import dedup, parse_datetime
 
 
 class TaskStatus(Enum):
@@ -91,9 +90,9 @@ def from_json(cls, response: dict[str, Any]) -> Self:
 class LaunchInfo:
     """Nextflow Tower workflow launch specification."""
 
-    compute_env_id: str
-    pipeline: str
-    work_dir: str
+    pipeline: Optional[str] = None
+    compute_env_id: Optional[str] = None
+    work_dir: Optional[str] = None
     revision: Optional[str] = None
     params: Optional[dict] = None
     nextflow_config: Optional[str] = None
@@ -104,18 +103,6 @@ class LaunchInfo:
     workspace_secrets: list[str] = field(default_factory=list)
     label_ids: list[int] = field(default_factory=list)
 
-    @staticmethod
-    def dedup(items: Collection[str]) -> list[str]:
-        """Deduplicate items in a collection.
-
-        Args:
-            items: Collection of items.
-
-        Returns:
-            Deduplicated collection or None.
-        """
-        return list(set(items))
-
     def fill_in(self, attr: str, value: Any):
         """Fill in any missing values.
 
@@ -126,6 +113,35 @@ def fill_in(self, attr: str, value: Any):
         if not getattr(self, attr, None):
             setattr(self, attr, value)
 
+    def add_in(self, attr: str, values: Iterable[Any]):
+        """Add values to a list attribute.
+
+        Args:
+            attr: Attribute name.
+            values: New attribute values.
+        """
+        current_values = getattr(self, attr)
+        if not isinstance(current_values, list):
+            message = f"Attribute '{attr}' is not a list and cannot be extended."
+            raise ValueError(message)
+        updated_values = current_values + list(values)
+        updated_values = dedup(updated_values)
+        setattr(self, attr, updated_values)
+
+    def get(self, name: str) -> Any:
+        """Retrieve attribute value, which cannot be None.
+
+        Args:
+            name: Atribute name.
+
+        Returns:
+            Attribute value (not None).
+        """
+        if getattr(self, name, None) is None:
+            message = f"Attribute '{name}' must be set (not None) by this point."
+            raise ValueError(message)
+        return getattr(self, name)
+
     def to_dict(self) -> dict[str, Any]:
         """Generate JSON representation of a launch specification.
 
@@ -134,19 +150,19 @@ def to_dict(self) -> dict[str, Any]:
         """
         output = {
             "launch": {
-                "computeEnvId": self.compute_env_id,
-                "configProfiles": self.dedup(self.profiles),
+                "computeEnvId": self.get("compute_env_id"),
+                "configProfiles": dedup(self.profiles),
                 "configText": self.nextflow_config,
                 "dateCreated": None,
                 "entryName": None,
                 "headJobCpus": None,
                 "headJobMemoryMb": None,
                 "id": None,
-                "labelIds": self.label_ids,
+                "labelIds": dedup(self.label_ids),
                 "mainScript": None,
                 "optimizationId": None,
                 "paramsText": json.dumps(self.params),
-                "pipeline": self.pipeline,
+                "pipeline": self.get("pipeline"),
                 "postRunScript": None,
                 "preRunScript": self.pre_run_script,
                 "pullLatest": False,
@@ -156,9 +172,9 @@ def to_dict(self) -> dict[str, Any]:
                 "schemaName": None,
                 "stubRun": False,
                 "towerConfig": None,
-                "userSecrets": self.dedup(self.user_secrets),
-                "workDir": self.work_dir,
-                "workspaceSecrets": self.dedup(self.workspace_secrets),
+                "userSecrets": dedup(self.user_secrets),
+                "workDir": self.get("work_dir"),
+                "workspaceSecrets": dedup(self.workspace_secrets),
             }
         }
         return output

src/orca/services/nextflowtower/ops.py

@@ -127,10 +127,11 @@ def launch_workflow(
         launch_info.fill_in("compute_env_id", compute_env_id)
         launch_info.fill_in("work_dir", compute_env.work_dir)
         launch_info.fill_in("pre_run_script", compute_env.pre_run_script)
-        launch_info.fill_in("label_ids", label_ids)
+        launch_info.add_in("label_ids", label_ids)
 
         return self.client.launch_workflow(launch_info, self.workspace_id)
 
+    # TODO: Consider switching return value to a namedtuple
     def get_workflow_status(self, workflow_id: str) -> tuple[TaskStatus, bool]:
         """Gets status of workflow run
 
@@ -146,5 +147,4 @@ def get_workflow_status(self, workflow_id: str) -> tuple[TaskStatus, bool]:
         )
         task_status = cast(TaskStatus, response["workflow"]["status"])
         is_done = task_status in TaskStatus.terminal_states.value
-        # TODO: Consider switching return value to a namedtuple
         return task_status, is_done

src/orca/services/nextflowtower/utils.py

@@ -1,4 +1,8 @@
+from collections.abc import Collection
 from datetime import datetime, timezone
+from typing import TypeVar
+
+T = TypeVar("T", int, str)
 
 
 def parse_datetime(text: str) -> datetime:
@@ -13,3 +17,15 @@ def parse_datetime(text: str) -> datetime:
     parsed = datetime.strptime(text, "%Y-%m-%dT%H:%M:%SZ")
     parsed = parsed.replace(tzinfo=timezone.utc)
     return parsed
+
+
+def dedup(items: Collection[T]) -> list[T]:
+    """Deduplicate items in a collection.
+
+    Args:
+        items: Collection of items.
+
+    Returns:
+        Deduplicated collection or None.
+    """
+    return list(set(items))

tests/services/nextflowtower/test_integration.py

@@ -39,12 +39,10 @@ def test_that_a_valid_client_can_be_constructed_and_tested(client):
 def test_that_a_workflow_can_be_launched(ops):
     scratch_bucket = "s3://orca-service-test-project-tower-scratch/"
     launch_info = models.LaunchInfo(
-        compute_env_id="5ykJFs33AE3d3AgThavz3b",
         pipeline="nf-core/rnaseq",
         revision="3.11.2",
         profiles=["test"],
         params={"outdir": f"{scratch_bucket}/2days/launch_test"},
-        work_dir=f"{scratch_bucket}/work",
     )
     workflow_id = ops.launch_workflow(launch_info, "ondemand")
     assert workflow_id

tests/services/nextflowtower/test_models.py

@@ -0,0 +1,34 @@
+import pytest
+
+from orca.services.nextflowtower.models import LaunchInfo
+
+
+def test_that_getting_an_launch_info_attribute_works():
+    launch_info = LaunchInfo(pipeline="foo")
+    assert launch_info.get("pipeline") == "foo"
+
+
+def test_for_an_error_when_getting_an_launch_info_attribute_that_is_missing():
+    launch_info = LaunchInfo()
+    with pytest.raises(ValueError):
+        launch_info.get("pipeline")
+
+
+def test_that_launch_info_attribute_can_be_filled_in():
+    launch_info = LaunchInfo()
+    assert launch_info.pipeline is None
+    launch_info.fill_in("pipeline", "foo")
+    assert launch_info.pipeline == "foo"
+
+
+def test_that_launch_info_list_attribute_can_be_added_in():
+    launch_info = LaunchInfo(label_ids=[1, 2, 3])
+    assert launch_info.label_ids == [1, 2, 3]
+    launch_info.add_in("label_ids", [4, 5, 6])
+    assert launch_info.label_ids == [1, 2, 3, 4, 5, 6]
+
+
+def test_for_an_error_when_adding_in_with_nonlist_launch_info_attribute():
+    launch_info = LaunchInfo(pipeline="foo")
+    with pytest.raises(ValueError):
+        launch_info.add_in("pipeline", [4, 5, 6])

tests/services/nextflowtower/test_utils.py

@@ -6,3 +6,9 @@
 def test_that_parse_datetime_works():
     result = utils.parse_datetime("2023-04-26T00:49:49Z")
     assert result == datetime(2023, 4, 26, 0, 49, 49, tzinfo=timezone.utc)
+
+
+def test_that_launch_info_dedup_works():
+    secrets = ["foo", "bar", "baz", "foo"]
+    dedupped = utils.dedup(secrets)
+    assert len(dedupped) == 3