Repository: fractal-analytics-platform/fractal-client (language: python)

Issue:
Expose the new workflow-subset-execution feature
Ref:
* https://github.com/fractal-analytics-platform/fractal-server/issues/783
* https://github.com/fractal-analytics-platform/fractal-server/issues/784
Expose the new workflow-subset-execution feature
Ref:
* https://github.com/fractal-analytics-platform/fractal-server/issues/783
* https://github.com/fractal-analytics-platform/fractal-server/issues/784


Discussion / hints:

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4c70daca0..e76297afa 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,6 @@
 **Note**: Numbers like (\#123) point to closed Pull Requests on the fractal repository.
 
+* Support `workflow apply --start/--end` arguments for submitting a workflow subset (\#527).
 * Exclude `common/tests` and other files from build (\#523).
 * Remove obsolete folders from `tests/data` (\#526).
 
diff --git a/fractal/cmd/__init__.py b/fractal/cmd/__init__.py
index 92dac5964..4a3436be4 100644
--- a/fractal/cmd/__init__.py
+++ b/fractal/cmd/__init__.py
@@ -254,6 +254,8 @@ async def workflow(
             "input_dataset_id",
             "output_dataset_id",
             "worker_init",
+            "first_task_index",
+            "last_task_index",
         ]
         function_kwargs = get_kwargs(parameters, kwargs)
         iface = await workflow_apply(client, **function_kwargs)
diff --git a/fractal/cmd/_workflow.py b/fractal/cmd/_workflow.py
index dc81b8f99..a6bb40586 100644
--- a/fractal/cmd/_workflow.py
+++ b/fractal/cmd/_workflow.py
@@ -226,6 +226,8 @@ async def workflow_apply(
     workflow_id: int,
     input_dataset_id: int,
     output_dataset_id: int,
+    first_task_index: Optional[int] = None,
+    last_task_index: Optional[int] = None,
     worker_init: Optional[str] = None,
 ) -> BaseInterface:
     apply_wf_create_dict = dict(
@@ -234,8 +236,12 @@ async def workflow_apply(
         output_dataset_id=output_dataset_id,
     )
     # Prepare ApplyWorkflowCreate object, without None attributes
-    if worker_init:
+    if worker_init is not None:
         apply_wf_create_dict["worker_init"] = worker_init
+    if first_task_index is not None:
+        apply_wf_create_dict["first_task_index"] = first_task_index
+    if last_task_index is not None:
+        apply_wf_create_dict["last_task_index"] = last_task_index
     apply_wf_create = ApplyWorkflowCreate(**apply_wf_create_dict)
 
     # Prepare query parameters
diff --git a/fractal/common b/fractal/common
index 2da510266..077c5925a 160000
--- a/fractal/common
+++ b/fractal/common
@@ -1,1 +1,1 @@
-Subproject commit 2da51026638291f410efdd9ac96193fe087d2cc0
+Subproject commit 077c5925afaf3bcb8618bfcbe5c54d482864c254
diff --git a/fractal/parser.py b/fractal/parser.py
index b81b8c408..871e6a988 100644
--- a/fractal/parser.py
+++ b/fractal/parser.py
@@ -656,10 +656,24 @@
     allow_abbrev=False,
 )
 
-workflow_apply_parser.add_argument("project_id")
-workflow_apply_parser.add_argument("workflow_id")
-workflow_apply_parser.add_argument("input_dataset_id")
-workflow_apply_parser.add_argument("output_dataset_id")
+workflow_apply_parser.add_argument("project_id", type=int)
+workflow_apply_parser.add_argument("workflow_id", type=int)
+workflow_apply_parser.add_argument("input_dataset_id", type=int)
+workflow_apply_parser.add_argument("output_dataset_id", type=int)
+workflow_apply_parser.add_argument(
+        "--start",
+        dest="first_task_index",
+        type=int,
+        help="Positional index of the first task to be executed",
+        required=False,
+        )
+workflow_apply_parser.add_argument(
+        "--end",
+        dest="last_task_index",
+        type=int,
+        help="Positional index of the last task to be executed",
+        required=False,
+        )
 workflow_apply_parser.add_argument(
     "-w",
     "--worker-init",
diff --git a/poetry.lock b/poetry.lock
index 9e7df8872..215577f00 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -604,7 +604,7 @@ sqlalchemy = ["fastapi-users-db-sqlalchemy (>=4.0.0)"]
 
 [[package]]
 name = "fractal-server"
-version = "1.3.3"
+version = "1.3.4"
 description = "Server component of the Fractal analytics platform"
 optional = false
 python-versions = "^3.9"
@@ -633,7 +633,7 @@ slurm = ["cloudpickle (>=2.2.1,<2.3.0)", "clusterfutures (>=0.5,<0.6)"]
 type = "git"
 url = "https://github.com/fractal-analytics-platform/fractal-server.git"
 reference = "main"
-resolved_reference = "04b72688129f64c257fa293d668f96b736548aee"
+resolved_reference = "239008e3e787e3a48155dc3ed67ae2b27a75ed79"
 
 [[package]]
 name = "greenlet"
diff --git a/pyproject.toml b/pyproject.toml
index 909a429fb..442df8296 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -16,6 +16,7 @@ license = "BSD-3-Clause"
 exclude = [
 "fractal/.gitignore",
 "fractal/common/tests",
+"fractal/common/requirements",
 "fractal/common/.git*",
 "fractal/common/.pre-commit-config.yaml",
 ]


Test patch:
diff --git a/tests/test_workflow.py b/tests/test_workflow.py
index 2a7a06ad9..40fba3e98 100644
--- a/tests/test_workflow.py
+++ b/tests/test_workflow.py
@@ -445,26 +445,30 @@ async def test_workflow_apply(
         )
         assert res.retcode == 0
 
-    # Create workflow and add task
+    # Create workflow and add task twice
     res = await invoke(f"workflow new {WORKFLOW_NAME} {prj_id}")
     workflow = res.data
     workflow_id = workflow["id"]
     debug(workflow)
     assert res.retcode == 0
-    TASK_ID = 1
-    res = await invoke(
-        f"workflow add-task {prj_id} {workflow_id} --task-id {TASK_ID}"
-    )
-    workflow_task = res.data
-    debug(workflow_task)
-    assert res.retcode == 0
-    TASK_NAME = res.data["task"]["name"]
-    debug(TASK_NAME)
+    for _ in [0, 1]:
+        TASK_ID = 1
+        res = await invoke(
+            f"workflow add-task {prj_id} {workflow_id} --task-id {TASK_ID}"
+        )
+        workflow_task = res.data
+        debug(workflow_task)
+        assert res.retcode == 0
+        TASK_NAME = res.data["task"]["name"]
+        debug(TASK_NAME)
 
     # Call `workflow apply`
+    FIRST_TASK_INDEX = 0
+    LAST_TASK_INDEX = 0
     cmd = (
         f"workflow apply "
-        f"{prj_id} {workflow_id} {input_dataset_id} {output_dataset_id}"
+        f"{prj_id} {workflow_id} {input_dataset_id} {output_dataset_id} "
+        f"--start {FIRST_TASK_INDEX} --end {LAST_TASK_INDEX}"
     )
     debug(cmd)
     res = await invoke(cmd)
@@ -474,6 +478,7 @@ async def test_workflow_apply(
     job_id = job["id"]
     assert job["status"] == "submitted"
 
+
     # Avoid immediately calling `job show` right after `workflow apply`
     time.sleep(1)
 
@@ -494,6 +499,11 @@ async def test_workflow_apply(
         assert time.perf_counter() - starting_time < TIMEOUT
     assert job["history"][0] == TASK_NAME
 
+    # Check that job has correct first_task_index and last_task_index
+    # attributes
+    assert job["first_task_index"] == FIRST_TASK_INDEX
+    assert job["last_task_index"] == LAST_TASK_INDEX
+
     # Prepare and run a workflow with a failing task
     args_file = str(tmp_path / "args.json")
     with open(args_file, "w") as f: