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

Issue:
Remove `rich`
As soon as the points listed below are closed, fractal-web should cover most (all) relevant uses of fractal-client, and meanwhile the web client is also much more used than the CLI one. I think we can simplify fractal-client even further, to make its maintenance almost trivial.

A first idea would be to move away from the nice rich-table-based display of data, also in view of the fact that it would be unreadable for larger data.

---

* https://github.com/fractal-analytics-platform/fractal-web/pull/336
* https://github.com/fractal-analytics-platform/fractal-web/issues/169
* https://github.com/fractal-analytics-platform/fractal-web/issues/342


Discussion / hints:

Fix patch:
diff --git a/fractal_client/client.py b/fractal_client/client.py
index b159c9b12..aafc0d735 100644
--- a/fractal_client/client.py
+++ b/fractal_client/client.py
@@ -25,7 +25,7 @@
 from .authclient import AuthClient
 from .authclient import AuthenticationError
 from .config import settings
-from .interface import PrintInterface
+from .interface import Interface
 from .parser import parser_main
 
 
@@ -112,9 +112,9 @@ def handle(cli_args: List[str] = argv):
             with AuthClient(username=username, password=password) as client:
                 interface = handler(client, **kwargs)
     except AuthenticationError as e:
-        return PrintInterface(retcode=1, data=e.args[0])
+        return Interface(retcode=1, data=e.args[0])
     except ConnectError as e:
-        return PrintInterface(
+        return Interface(
             retcode=1,
             data=(
                 f"ConnectError at {e.request.url}\n"
@@ -123,7 +123,7 @@ def handle(cli_args: List[str] = argv):
             ),
         )
     except Exception as e:
-        return PrintInterface(retcode=1, data=str(e))
+        return Interface(retcode=1, data=str(e))
 
     return interface
 
diff --git a/fractal_client/cmd/__init__.py b/fractal_client/cmd/__init__.py
index 8fa364495..ec633c3ef 100644
--- a/fractal_client/cmd/__init__.py
+++ b/fractal_client/cmd/__init__.py
@@ -2,19 +2,15 @@
 
 from ..authclient import AuthClient
 from ..config import settings
-from ..interface import BaseInterface
-from ..interface import PrintInterface
+from ..interface import Interface
 from ._dataset import delete_dataset
-from ._dataset import delete_resource
 from ._dataset import get_dataset
-from ._dataset import get_dataset_history
-from ._dataset import get_dataset_status
 from ._dataset import patch_dataset
 from ._dataset import post_dataset
-from ._dataset import post_resource
 from ._job import get_job
 from ._job import get_job_list
 from ._job import get_job_logs
+from ._job import job_submit
 from ._job import stop_job
 from ._project import delete_project
 from ._project import get_project
@@ -40,9 +36,6 @@
 from ._workflow import patch_workflowtask
 from ._workflow import post_workflow
 from ._workflow import post_workflowtask
-from ._workflow import workflow_apply
-from ._workflow import workflow_export
-from ._workflow import workflow_import
 from fractal_client import __VERSION__
 
 
@@ -59,7 +52,7 @@ def project(
     subcmd: str,
     batch: bool = False,
     **kwargs,
-) -> BaseInterface:
+) -> Interface:
 
     if subcmd == "new":
         parameters = ["name"]
@@ -75,19 +68,11 @@ def project(
         parameters = [
             "project_id",
             "new_name",
-            "make_read_only",
-            "remove_read_only",
         ]
         function_kwargs = get_kwargs(parameters, kwargs)
         iface = patch_project(client, **function_kwargs)
     elif subcmd == "add-dataset":
-        parameters = [
-            "project_id",
-            "dataset_name",
-            "metadata",
-            "type",
-            "make_read_only",
-        ]
+        parameters = ["project_id", "dataset_name", "zarr_dir", "filters"]
         function_kwargs = get_kwargs(parameters, kwargs)
         iface = post_dataset(client, batch=batch, **function_kwargs)
     elif subcmd == "delete":
@@ -105,28 +90,17 @@ def dataset(
     subcmd: str,
     batch: bool = False,
     **kwargs,
-) -> BaseInterface:
+) -> Interface:
     if subcmd == "show":
         parameters = ["project_id", "dataset_id"]
         function_kwargs = get_kwargs(parameters, kwargs)
         iface = get_dataset(client, **function_kwargs)
-    elif subcmd == "add-resource":
-        parameters = ["project_id", "dataset_id", "path"]
-        function_kwargs = get_kwargs(parameters, kwargs)
-        iface = post_resource(client, batch=batch, **function_kwargs)
-    elif subcmd == "rm-resource":
-        parameters = ["project_id", "dataset_id", "resource_id"]
-        function_kwargs = get_kwargs(parameters, kwargs)
-        iface = delete_resource(client, **function_kwargs)
     elif subcmd == "edit":
         parameters = [
             "project_id",
             "dataset_id",
             "new_name",
-            "new_type",
-            "meta_file",
-            "make_read_only",
-            "remove_read_only",
+            "filters",
         ]
         function_kwargs = get_kwargs(parameters, kwargs)
         iface = patch_dataset(client, **function_kwargs)
@@ -134,14 +108,6 @@ def dataset(
         parameters = ["project_id", "dataset_id"]
         function_kwargs = get_kwargs(parameters, kwargs)
         iface = delete_dataset(client, **function_kwargs)
-    elif subcmd == "history":
-        parameters = ["project_id", "dataset_id"]
-        function_kwargs = get_kwargs(parameters, kwargs)
-        iface = get_dataset_history(client, **function_kwargs)
-    elif subcmd == "status":
-        parameters = ["project_id", "dataset_id"]
-        function_kwargs = get_kwargs(parameters, kwargs)
-        iface = get_dataset_status(client, **function_kwargs)
     else:
         raise NoCommandError(f"Command dataset {subcmd} not found")
     return iface
@@ -152,7 +118,7 @@ def task(
     subcmd: str,
     batch: bool = False,
     **kwargs,
-) -> BaseInterface:
+) -> Interface:
 
     if subcmd == "list":
         iface = get_task_list(client)
@@ -214,7 +180,7 @@ def workflow(
     subcmd: str,
     batch: bool = False,
     **kwargs,
-) -> BaseInterface:
+) -> Interface:
     if subcmd == "show":
         parameters = ["project_id", "workflow_id"]
         function_kwargs = get_kwargs(parameters, kwargs)
@@ -243,8 +209,11 @@ def workflow(
             "task_name",
             "task_version",
             "order",
-            "args_file",
-            "meta_file",
+            "args_non_parallel",
+            "args_parallel",
+            "input_filters",
+            "meta_parallel",
+            "meta_non_parallel",
         ]
         function_kwargs = get_kwargs(parameters, kwargs)
         iface = post_workflowtask(client, batch=batch, **function_kwargs)
@@ -253,8 +222,11 @@ def workflow(
             "project_id",
             "workflow_id",
             "workflow_task_id",
-            "args_file",
-            "meta_file",
+            "input_filters",
+            "args_non_parallel",
+            "args_parallel",
+            "meta_parallel",
+            "meta_non_parallel",
         ]
         function_kwargs = get_kwargs(parameters, kwargs)
         iface = patch_workflowtask(client, **function_kwargs)
@@ -262,26 +234,6 @@ def workflow(
         parameters = ["project_id", "workflow_id", "workflow_task_id"]
         function_kwargs = get_kwargs(parameters, kwargs)
         iface = delete_workflowtask(client, **function_kwargs)
-    elif subcmd == "apply":
-        parameters = [
-            "project_id",
-            "workflow_id",
-            "input_dataset_id",
-            "output_dataset_id",
-            "worker_init",
-            "first_task_index",
-            "last_task_index",
-        ]
-        function_kwargs = get_kwargs(parameters, kwargs)
-        iface = workflow_apply(client, batch=batch, **function_kwargs)
-    elif subcmd == "import":
-        parameters = ["project_id", "json_file"]
-        function_kwargs = get_kwargs(parameters, kwargs)
-        iface = workflow_import(client, batch=batch, **function_kwargs)
-    elif subcmd == "export":
-        parameters = ["project_id", "workflow_id", "json_file"]
-        function_kwargs = get_kwargs(parameters, kwargs)
-        iface = workflow_export(client, **function_kwargs)
     else:
         raise NoCommandError(f"Command workflow {subcmd} not found")
     return iface
@@ -292,7 +244,7 @@ def job(
     subcmd: str,
     batch: bool = False,
     **kwargs,
-) -> BaseInterface:
+) -> Interface:
     if subcmd == "list":
         parameters = ["project_id"]
         function_kwargs = get_kwargs(parameters, kwargs)
@@ -309,16 +261,27 @@ def job(
         parameters = ["project_id", "job_id"]
         function_kwargs = get_kwargs(parameters, kwargs)
         iface = stop_job(client, **function_kwargs)
+    elif subcmd == "submit":
+        parameters = [
+            "project_id",
+            "workflow_id",
+            "dataset_id",
+            "worker_init",
+            "first_task_index",
+            "last_task_index",
+        ]
+        function_kwargs = get_kwargs(parameters, kwargs)
+        iface = job_submit(client, batch=batch, **function_kwargs)
     else:
         raise NoCommandError(f"Command job {subcmd} not found")
     return iface
 
 
-def version(client: Client, **kwargs) -> PrintInterface:
+def version(client: Client, **kwargs) -> Interface:
     res = client.get(f"{settings.FRACTAL_SERVER}/api/alive/")
     data = res.json()
 
-    return PrintInterface(
+    return Interface(
         retcode=0,
         data=(
             f"Fractal client\n\tversion: {__VERSION__}\n"
@@ -331,7 +294,7 @@ def version(client: Client, **kwargs) -> PrintInterface:
 
 def user(
     client: AuthClient, subcmd: str, batch: bool = False, **kwargs
-) -> BaseInterface:
+) -> Interface:
     if subcmd == "register":
         parameters = [
             "new_email",
diff --git a/fractal_client/cmd/_dataset.py b/fractal_client/cmd/_dataset.py
index e3220788c..8b60b392f 100644
--- a/fractal_client/cmd/_dataset.py
+++ b/fractal_client/cmd/_dataset.py
@@ -1,14 +1,9 @@
 import json
 from typing import Optional
 
-from rich.table import Table
-
 from ..authclient import AuthClient
 from ..config import settings
-from ..interface import BaseInterface
-from ..interface import PrintInterface
-from ..interface import RichConsoleInterface
-from ..interface import RichJsonInterface
+from ..interface import Interface
 from ..response import check_response
 
 
@@ -17,28 +12,25 @@ def post_dataset(
     *,
     project_id: int,
     dataset_name: str,
-    metadata: Optional[str] = None,
-    type: Optional[str] = None,
+    zarr_dir: str,
+    filters: Optional[str] = None,
     batch: bool = False,
-    make_read_only: bool = False,
-) -> RichJsonInterface:
+) -> Interface:
     """
     Arguments:
         project_id: ID of project to add the new dataset to
         dataset_name: Name of new dataset
-        metadata: Path to file containing dataset metadata in JSON format.
-        type: Dataset type.
-        make_read_only: Make the new dataset read-only.
+        filters: Path to file containing dataset filters in JSON format.
+        batch: Dataset filters.
     """
-    if metadata is None:
-        meta = {}
+    if filters is None:
+        dataset = dict(name=dataset_name, zarr_dir=zarr_dir)
     else:
-        with open(metadata, "r") as f:
-            meta = json.load(f)
-
-    dataset = dict(name=dataset_name, meta=meta, read_only=make_read_only)
-    if type:
-        dataset["type"] = type
+        with open(filters, "r") as f:
+            filters_dict = json.load(f)
+        dataset = dict(
+            name=dataset_name, filters=filters_dict, zarr_dir=zarr_dir
+        )
 
     res = client.post(
         f"{settings.BASE_URL}/project/{project_id}/dataset/",
@@ -46,49 +38,9 @@ def post_dataset(
     )
     new_dataset = check_response(res, expected_status_code=201)
     if batch:
-        return PrintInterface(retcode=0, data=new_dataset["id"])
-    else:
-        return RichJsonInterface(retcode=0, data=new_dataset)
-
-
-def post_resource(
-    client: AuthClient,
-    *,
-    project_id: int,
-    dataset_id: int,
-    path: str,
-    batch: bool = False,
-) -> BaseInterface:
-
-    res = client.post(
-        (
-            f"{settings.BASE_URL}/project/{project_id}/"
-            f"dataset/{dataset_id}/resource/"
-        ),
-        json=dict(path=path),
-    )
-    new_resource = check_response(res, expected_status_code=201)
-    if batch:
-        return PrintInterface(retcode=0, data=new_resource["id"])
+        return Interface(retcode=0, data=new_dataset["id"])
     else:
-        return RichJsonInterface(retcode=0, data=new_resource)
-
-
-def delete_resource(
-    client: AuthClient,
-    *,
-    project_id: int,
-    dataset_id: int,
-    resource_id: int,
-) -> BaseInterface:
-    res = client.delete(
-        (
-            f"{settings.BASE_URL}/project/{project_id}/"
-            f"dataset/{dataset_id}/resource/{resource_id}/"
-        )
-    )
-    check_response(res, expected_status_code=204)
-    return PrintInterface(retcode=0, data="")
+        return Interface(retcode=0, data=new_dataset)
 
 
 def patch_dataset(
@@ -97,29 +49,16 @@ def patch_dataset(
     project_id: int,
     dataset_id: int,
     new_name: Optional[str] = None,
-    new_type: Optional[str] = None,
-    meta_file: Optional[str] = None,
-    make_read_only: bool = False,
-    remove_read_only: bool = False,
-) -> BaseInterface:
-
+    filters: Optional[str] = None,
+) -> Interface:
     # Prepare payload
     dataset_update = {}
     if new_name:
         dataset_update["name"] = new_name
-    if new_type:
-        dataset_update["type"] = new_type
-    if meta_file:
-        with open(meta_file, "r") as f:
-            meta = json.load(f)
-        dataset_update.update(meta=meta)
-    if make_read_only:
-        dataset_update["read_only"] = True
-    if remove_read_only:
-        dataset_update["read_only"] = False
-
-    if not dataset_update:
-        return PrintInterface(retcode=1, data="Nothing to update")
+    if filters:
+        with open(filters, "r") as f:
+            filters_from_file = json.load(f)
+        dataset_update["filters"] = filters_from_file
 
     res = client.patch(
         (
@@ -129,71 +68,24 @@ def patch_dataset(
         json=dataset_update,
     )
     data = check_response(res, expected_status_code=200)
-    return RichJsonInterface(retcode=0, data=data)
+    return Interface(retcode=0, data=data)
 
 
 def get_dataset(
     client: AuthClient, *, project_id: int, dataset_id: int
-) -> BaseInterface:
+) -> Interface:
     res = client.get(
         f"{settings.BASE_URL}/project/{project_id}/dataset/{dataset_id}/"
     )
-    from rich.console import Group
-
-    dataset = check_response(res, expected_status_code=200)
-
-    table = Table(title="Dataset")
-    table.add_column("ID", style="cyan", no_wrap=True)
-    table.add_column("Name", justify="right", style="green")
-    table.add_column("Type", style="white")
-    table.add_column("Meta", justify="center")
-    table.add_column("Read only", justify="center")
-
-    table.add_row(
-        str(dataset["id"]),
-        dataset["name"],
-        dataset["type"],
-        json.dumps(dataset["meta"], indent=2),
-        "✅" if dataset["read_only"] else "❌",
-    )
-    table_res = Table(title="Resources")
-    table_res.add_column("Path", justify="center", style="yellow")
-    table_res.add_column("ID", justify="center", style="yellow")
-    table_res.add_column("Dataset ID", justify="center", style="yellow")
-    for r in dataset["resource_list"]:
-        table_res.add_row(r["path"], str(r["id"]), str(r["dataset_id"]))
-    group = Group(table, table_res)
-    return RichConsoleInterface(retcode=0, data=group)
+    return Interface(retcode=0, data=res.json())
 
 
 def delete_dataset(
     client: AuthClient, *, project_id: int, dataset_id: int
-) -> PrintInterface:
+) -> Interface:
 
     res = client.delete(
         f"{settings.BASE_URL}/project/{project_id}/dataset/{dataset_id}/"
     )
     check_response(res, expected_status_code=204)
-    return PrintInterface(retcode=0, data="")
-
-
-def get_dataset_history(
-    client: AuthClient, *, project_id: int, dataset_id: int
-) -> BaseInterface:
-    res = client.get(
-        f"{settings.BASE_URL}/project/{project_id}/dataset/{dataset_id}/"
-        "export_history/"
-    )
-    history_workflow = check_response(res, expected_status_code=200)
-    return RichJsonInterface(retcode=0, data=history_workflow)
-
-
-def get_dataset_status(
-    client: AuthClient, *, project_id: int, dataset_id: int
-) -> BaseInterface:
-    res = client.get(
-        f"{settings.BASE_URL}/project/{project_id}/dataset/{dataset_id}/"
-        "status/"
-    )
-    dataset_status = check_response(res, expected_status_code=200)
-    return RichJsonInterface(retcode=0, data=dataset_status)
+    return Interface(retcode=0, data="")
diff --git a/fractal_client/cmd/_job.py b/fractal_client/cmd/_job.py
index ea1adc265..14cc836ce 100644
--- a/fractal_client/cmd/_job.py
+++ b/fractal_client/cmd/_job.py
@@ -1,16 +1,12 @@
 import logging
 import os
 from pathlib import Path
+from typing import Optional
 from zipfile import ZipFile
 
-from rich.table import Table
-
 from ..authclient import AuthClient
 from ..config import settings
-from ..interface import BaseInterface
-from ..interface import PrintInterface
-from ..interface import RichConsoleInterface
-from ..interface import RichJsonInterface
+from ..interface import Interface
 from ..response import check_response
 
 
@@ -19,9 +15,8 @@ def get_job(
     *,
     project_id: int,
     job_id: int,
-    do_not_separate_logs: bool = False,
     batch: bool = False,
-) -> BaseInterface:
+) -> Interface:
     """
     Query the status of a workflow-execution job
     """
@@ -29,47 +24,23 @@ def get_job(
     res = client.get(f"{settings.BASE_URL}/project/{project_id}/job/{job_id}/")
     job = check_response(res, expected_status_code=200)
     if batch:
-        return PrintInterface(retcode=0, data=job["status"])
+        return Interface(retcode=0, data=job["status"])
     else:
-        if do_not_separate_logs or (job.get("log") is None):
-            return RichJsonInterface(retcode=0, data=job)
-        else:
-            log = job.pop("log")
-            extra_lines = ["\nThis is the job log:\n", log]
-            return RichJsonInterface(
-                retcode=0, data=job, extra_lines=extra_lines
-            )
+        return Interface(retcode=0, data=job)
 
 
 def get_job_list(
     client: AuthClient, *, project_id: int, batch: bool = False
-) -> BaseInterface:
+) -> Interface:
 
     res = client.get(f"{settings.BASE_URL}/project/{project_id}/job/")
     jobs = check_response(res, expected_status_code=200)
 
     if batch:
         job_ids = " ".join(str(job["id"]) for job in jobs)
-        return PrintInterface(retcode=0, data=job_ids)
+        return Interface(retcode=0, data=job_ids)
     else:
-        table = Table(title=f"Job list for project {project_id}")
-        kwargs = dict(style="white", justify="center")
-        table.add_column("id", **kwargs)
-        table.add_column("start_timestamp", **kwargs)
-        table.add_column("status", **kwargs)
-        table.add_column("workflow_id", **kwargs)
-        table.add_column("working_dir", **kwargs)
-
-        for j in jobs:
-            timestamp = j["start_timestamp"]
-            table.add_row(
-                str(j["id"]),
-                timestamp,
-                j["status"],
-                str(j["workflow_id"]),
-                j["working_dir"],
-            )
-        return RichConsoleInterface(retcode=0, data=table)
+        return Interface(retcode=0, data=jobs)
 
 
 def get_job_logs(
@@ -78,11 +49,11 @@ def get_job_logs(
     project_id: int,
     job_id: int,
     output_folder: str,
-) -> BaseInterface:
+) -> Interface:
 
     # Check that output_folder does not already exist
     if Path(output_folder).exists():
-        return PrintInterface(
+        return Interface(
             retcode=1, data=f"ERROR: {output_folder=} already exists"
         )
 
@@ -135,14 +106,10 @@ def get_job_logs(
     # Remove zipped temporary file
     os.unlink(zipped_archive_path)
 
-    return PrintInterface(
-        retcode=0, data=f"Logs downloaded to {output_folder=}"
-    )
+    return Interface(retcode=0, data=f"Logs downloaded to {output_folder=}")
 
 
-def stop_job(
-    client: AuthClient, *, project_id: int, job_id: int
-) -> BaseInterface:
+def stop_job(client: AuthClient, *, project_id: int, job_id: int) -> Interface:
     """
     Stop a workflow-execution job
     """
@@ -151,6 +118,45 @@ def stop_job(
         f"{settings.BASE_URL}/project/{project_id}/job/{job_id}/stop/"
     )
     check_response(res, expected_status_code=202)
-    return PrintInterface(
+    return Interface(
         retcode=0, data="Correctly called the job-stopping endpoint"
     )
+
+
+def job_submit(
+    client: AuthClient,
+    *,
+    project_id: int,
+    workflow_id: int,
+    dataset_id: int,
+    first_task_index: Optional[int] = None,
+    last_task_index: Optional[int] = None,
+    worker_init: Optional[str] = None,
+    batch: bool = False,
+) -> Interface:
+
+    job_submit = dict()
+    # Prepare JobV2 object, without None attributes
+    if worker_init is not None:
+        job_submit["worker_init"] = worker_init
+    if first_task_index is not None:
+        job_submit["first_task_index"] = first_task_index
+    if last_task_index is not None:
+        job_submit["last_task_index"] = last_task_index
+
+    # Prepare query parameters
+    query_parameters = f"workflow_id={workflow_id}" f"&dataset_id={dataset_id}"
+
+    res = client.post(
+        (
+            f"{settings.BASE_URL}/project/{project_id}/job/"
+            f"submit/?{query_parameters}"
+        ),
+        json=job_submit,
+    )
+    job_read = check_response(res, expected_status_code=202)
+
+    if batch:
+        return Interface(retcode=0, data=job_read["id"])
+    else:
+        return Interface(retcode=0, data=job_read)
diff --git a/fractal_client/cmd/_project.py b/fractal_client/cmd/_project.py
index d084290e4..067df1d26 100644
--- a/fractal_client/cmd/_project.py
+++ b/fractal_client/cmd/_project.py
@@ -1,14 +1,8 @@
 from typing import Optional
-from typing import Union
-
-from rich.table import Table
 
 from ..authclient import AuthClient
 from ..config import settings
-from ..interface import BaseInterface
-from ..interface import PrintInterface
-from ..interface import RichConsoleInterface
-from ..interface import RichJsonInterface
+from ..interface import Interface
 from ..response import check_response
 
 
@@ -17,55 +11,36 @@ def post_project(
     *,
     name: str,
     batch: bool = False,
-) -> BaseInterface:
+) -> Interface:
     # Prepare a ProjectCreate request body
     payload = dict(name=name)
     # Send API request
     res = client.post(f"{settings.BASE_URL}/project/", json=payload)
     project = check_response(res, expected_status_code=201)
     if batch:
-        return PrintInterface(retcode=0, data=project["id"])
+        return Interface(retcode=0, data=project["id"])
     else:
-        return RichJsonInterface(retcode=0, data=project)
+        return Interface(retcode=0, data=project)
 
 
-def get_project_list(client: AuthClient) -> RichConsoleInterface:
+def get_project_list(client: AuthClient) -> Interface:
 
     res = client.get(f"{settings.BASE_URL}/project/")
     projects = check_response(res, expected_status_code=200)
+    return Interface(retcode=0, data=projects)
 
-    table = Table(title="Project List")
-    table.add_column("ID", style="cyan", no_wrap=True)
-    table.add_column("Name", style="magenta")
-    table.add_column("Read only", justify="center")
-
-    for p in projects:
-        # Map p.read_only (True/False) to read_only_icon (✅/❌)
-        if p["read_only"]:
-            read_only_icon = "✅"
-        else:
-            read_only_icon = "❌"
-
-        table.add_row(
-            str(p["id"]),
-            p["name"],
-            read_only_icon,
-        )
 
-    return RichConsoleInterface(retcode=0, data=table)
-
-
-def get_project(client: AuthClient, *, project_id: int) -> RichJsonInterface:
+def get_project(client: AuthClient, *, project_id: int) -> Interface:
     res = client.get(f"{settings.BASE_URL}/project/{project_id}/")
     project = check_response(res, expected_status_code=200)
-    return RichJsonInterface(retcode=0, data=project)
+    return Interface(retcode=0, data=project)
 
 
-def delete_project(client: AuthClient, *, project_id: int) -> PrintInterface:
+def delete_project(client: AuthClient, *, project_id: int) -> Interface:
 
     res = client.delete(f"{settings.BASE_URL}/project/{project_id}/")
     check_response(res, expected_status_code=204)
-    return PrintInterface(retcode=0, data="")
+    return Interface(retcode=0, data="")
 
 
 def patch_project(
@@ -73,22 +48,13 @@ def patch_project(
     *,
     project_id: int,
     new_name: Optional[str] = None,
-    make_read_only: bool = False,
-    remove_read_only: bool = False,
-) -> Union[RichJsonInterface, PrintInterface]:
+) -> Interface:
     project_update = {}
     if new_name:
         project_update["name"] = new_name
-    if make_read_only:
-        project_update["read_only"] = True
-    if remove_read_only:
-        project_update["read_only"] = False
-
-    if not project_update:
-        return PrintInterface(retcode=1, data="Nothing to update")
 
     res = client.patch(
         f"{settings.BASE_URL}/project/{project_id}/", json=project_update
     )
     new_project = check_response(res, expected_status_code=200)
-    return RichJsonInterface(retcode=0, data=new_project)
+    return Interface(retcode=0, data=new_project)
diff --git a/fractal_client/cmd/_task.py b/fractal_client/cmd/_task.py
index 05d102e6e..328d69342 100644
--- a/fractal_client/cmd/_task.py
+++ b/fractal_client/cmd/_task.py
@@ -1,22 +1,17 @@
-import json
 import logging
 import sys
 from typing import Optional
 
 from ..authclient import AuthClient
 from ..config import settings
-from ..interface import BaseInterface
-from ..interface import PrintInterface
-from ..interface import RichJsonInterface
+from ..interface import Interface
 from ..response import check_response
-from ._aux_task_caching import FractalCacheError
-from ._aux_task_caching import get_task_id_from_cache
 from ._aux_task_caching import refresh_task_cache
 
 
-def get_task_list(client: AuthClient) -> RichJsonInterface:
+def get_task_list(client: AuthClient) -> Interface:
     task_list = refresh_task_cache(client=client)
-    return RichJsonInterface(retcode=0, data=task_list)
+    return Interface(retcode=0, data=task_list)
 
 
 def task_collect_pip(
@@ -28,7 +23,7 @@ def task_collect_pip(
     package_extras: Optional[str] = None,
     pinned_dependency: Optional[list[str]] = None,
     batch: bool = False,
-) -> BaseInterface:
+) -> Interface:
 
     # Construct TaskCollectPip object
     task_collect = dict(package=package)
@@ -58,9 +53,9 @@ def task_collect_pip(
     state = check_response(res, expected_status_code=[200, 201])
     if batch:
         output = f"{state['id']} {state['data']['venv_path']}"
-        return PrintInterface(retcode=0, data=output)
+        return Interface(retcode=0, data=output)
     else:
-        return RichJsonInterface(retcode=0, data=state)
+        return Interface(retcode=0, data=state)
 
 
 def task_collection_check(
@@ -68,8 +63,7 @@ def task_collection_check(
     *,
     state_id: int,
     include_logs: bool,
-    do_not_separate_logs: bool = False,
-) -> BaseInterface:
+) -> Interface:
 
     res = client.get(
         f"{settings.BASE_URL}/task/collect/{state_id}/?verbose={include_logs}"
@@ -79,138 +73,16 @@ def task_collection_check(
     # Remove key-value pairs with None value
     state["data"] = {key: val for (key, val) in state["data"].items() if val}
 
-    if (not include_logs) or do_not_separate_logs:
-        return RichJsonInterface(retcode=0, data=state)
-    else:
-        log = state["data"].pop("log")
-        extra_lines = ["\nThis is the task-collection log:\n", log]
-        return RichJsonInterface(
-            retcode=0, data=state, extra_lines=extra_lines
-        )
+    return Interface(retcode=0, data=state)
 
 
-def post_task(
-    client: AuthClient,
-    *,
-    name: str,
-    command: str,
-    source: str,
-    input_type: str = "Any",
-    output_type: str = "Any",
-    batch: bool = False,
-    version: Optional[str] = None,
-    meta_file: Optional[str] = None,
-    args_schema: Optional[str] = None,
-    args_schema_version: Optional[str] = None,
-) -> BaseInterface:
-    task = dict(
-        name=name,
-        command=command,
-        source=source,
-        input_type=input_type,
-        output_type=output_type,
-    )
-    if version:
-        task["version"] = version
-    if meta_file:
-        with open(meta_file, "r") as f:
-            task["meta"] = json.load(f)
-    if args_schema:
-        with open(args_schema, "r") as f:
-            task["args_schema"] = json.load(f)
-    if args_schema_version:
-        task["args_schema_version"] = args_schema_version
-
-    res = client.post(f"{settings.BASE_URL}/task/", json=task)
-    new_task = check_response(res, expected_status_code=201)
+def post_task() -> None:
+    raise NotImplementedError
 
-    if batch:
-        return PrintInterface(retcode=0, data=str(new_task["id"]))
-    else:
-        return RichJsonInterface(retcode=0, data=new_task)
 
+def patch_task() -> None:
+    raise NotImplementedError
 
-def patch_task(
-    client: AuthClient,
-    *,
-    id: Optional[int] = None,
-    name: Optional[str] = None,
-    version: Optional[str] = None,
-    new_name: Optional[str] = None,
-    new_command: Optional[str] = None,
-    new_input_type: Optional[str] = None,
-    new_output_type: Optional[str] = None,
-    new_version: Optional[str] = None,
-    meta_file: Optional[str] = None,
-    new_args_schema: Optional[str] = None,
-    new_args_schema_version: Optional[str] = None,
-) -> BaseInterface:
-
-    if id:
-        if version:
-            logging.error(
-                "Too many arguments: cannot provide both `id` and `version`."
-            )
-            sys.exit(1)
-    else:
-        try:
-            id = get_task_id_from_cache(
-                client=client, task_name=name, version=version
-            )
-        except FractalCacheError as e:
-            print(e)
-            sys.exit(1)
-
-    task_update = {}
-    if new_name:
-        task_update["name"] = new_name
-    if new_input_type:
-        task_update["input_type"] = new_input_type
-    if new_output_type:
-        task_update["output_type"] = new_output_type
-    if new_command:
-        task_update["command"] = new_command
-    if new_version:
-        task_update["version"] = new_version
-    if meta_file:
-        with open(meta_file, "r") as f:
-            task_update["meta"] = json.load(f)
-    if new_args_schema:
-        with open(new_args_schema, "r") as f:
-            task_update["args_schema"] = json.load(f)
-    if new_args_schema_version:
-        task_update["args_schema_version"] = new_args_schema_version
-
-    if not task_update:
-        return PrintInterface(retcode=1, data="Nothing to update")
-
-    res = client.patch(f"{settings.BASE_URL}/task/{id}/", json=task_update)
-    new_task = check_response(res, expected_status_code=200)
-    return RichJsonInterface(retcode=0, data=new_task)
-
-
-def delete_task(
-    client: AuthClient,
-    *,
-    id: Optional[int] = None,
-    name: Optional[str] = None,
-    version: Optional[str] = None,
-) -> PrintInterface:
-
-    if id:
-        if version:
-            logging.error(
-                "Too many arguments: cannot provide both `id` and `version`."
-            )
-            sys.exit(1)
-    else:
-        try:
-            id = get_task_id_from_cache(
-                client=client, task_name=name, version=version
-            )
-        except FractalCacheError as e:
-            print(e)
-            sys.exit(1)
-    res = client.delete(f"{settings.BASE_URL}/task/{id}/")
-    check_response(res, expected_status_code=204)
-    return PrintInterface(retcode=0, data="")
+
+def delete_task() -> None:
+    raise NotImplementedError
diff --git a/fractal_client/cmd/_user.py b/fractal_client/cmd/_user.py
index 17b55492f..269f45ef8 100644
--- a/fractal_client/cmd/_user.py
+++ b/fractal_client/cmd/_user.py
@@ -1,10 +1,8 @@
 from typing import Optional
-from typing import Union
 
 from ..authclient import AuthClient
 from ..config import settings
-from ..interface import PrintInterface
-from ..interface import RichJsonInterface
+from ..interface import Interface
 from ..response import check_response
 
 
@@ -19,7 +17,7 @@ def user_register(
     superuser: bool = False,
     verified: bool = True,  # TODO: this is not currently exposed in the CLI
     batch: bool = False,
-) -> Union[RichJsonInterface, PrintInterface]:
+) -> Interface:
 
     new_user = dict(
         email=new_email,
@@ -40,7 +38,7 @@ def user_register(
         if new_password == confirm_new_password:
             new_user["password"] = new_password
         else:
-            return PrintInterface(retcode=1, data="Passwords do not match.")
+            return Interface(retcode=1, data="Passwords do not match.")
 
     res = client.post(
         f"{settings.FRACTAL_SERVER}/auth/register/", json=new_user
@@ -57,21 +55,21 @@ def user_register(
         data = check_response(res, expected_status_code=200)
 
     if batch:
-        return PrintInterface(retcode=0, data=data["id"])
+        return Interface(retcode=0, data=data["id"])
     else:
-        return RichJsonInterface(retcode=0, data=data)
+        return Interface(retcode=0, data=data)
 
 
-def user_list(client: AuthClient) -> RichJsonInterface:
+def user_list(client: AuthClient) -> Interface:
     res = client.get(f"{settings.FRACTAL_SERVER}/auth/users/")
     users = check_response(res, expected_status_code=200)
-    return RichJsonInterface(retcode=0, data=users)
+    return Interface(retcode=0, data=users)
 
 
-def user_show(client: AuthClient, *, user_id: str) -> RichJsonInterface:
+def user_show(client: AuthClient, *, user_id: str) -> Interface:
     res = client.get(f"{settings.FRACTAL_SERVER}/auth/users/{user_id}/")
     user = check_response(res, expected_status_code=200)
-    return RichJsonInterface(retcode=0, data=user)
+    return Interface(retcode=0, data=user)
 
 
 def user_edit(
@@ -87,7 +85,7 @@ def user_edit(
     remove_superuser: bool = False,
     make_verified: bool = False,
     remove_verified: bool = False,
-) -> Union[RichJsonInterface, PrintInterface]:
+) -> Interface:
 
     user_update = dict()
     if new_email is not None:
@@ -95,7 +93,7 @@ def user_edit(
             # Since `fastapi-users` sets `is_verified` to `False` each time the
             # email is updated, we force the user to make explicit whether the
             # account is verified or not.
-            return PrintInterface(
+            return Interface(
                 retcode=1,
                 data=(
                     "Cannot use `--new-email` without `--make-verified` or "
@@ -120,9 +118,6 @@ def user_edit(
     if new_username is not None:
         user_update["username"] = new_username
 
-    if not user_update:
-        return PrintInterface(retcode=1, data="Nothing to update")
-
     res = client.patch(
         f"{settings.FRACTAL_SERVER}/auth/users/{user_id}/", json=user_update
     )
@@ -137,15 +132,13 @@ def user_edit(
         )
         new_user = check_response(res, expected_status_code=200)
 
-    return RichJsonInterface(retcode=0, data=new_user)
+    return Interface(retcode=0, data=new_user)
 
 
-def user_whoami(
-    client: AuthClient, *, batch: bool
-) -> Union[RichJsonInterface, PrintInterface]:
+def user_whoami(client: AuthClient, *, batch: bool) -> Interface:
     res = client.get(f"{settings.FRACTAL_SERVER}/auth/current-user/")
     user = check_response(res, expected_status_code=200)
     if batch:
-        return PrintInterface(retcode=0, data=user["id"])
+        return Interface(retcode=0, data=user["id"])
     else:
-        return RichJsonInterface(retcode=0, data=user)
+        return Interface(retcode=0, data=user)
diff --git a/fractal_client/cmd/_workflow.py b/fractal_client/cmd/_workflow.py
index 214aa3207..f03d1cd11 100644
--- a/fractal_client/cmd/_workflow.py
+++ b/fractal_client/cmd/_workflow.py
@@ -6,9 +6,7 @@
 
 from ..authclient import AuthClient
 from ..config import settings
-from ..interface import BaseInterface
-from ..interface import PrintInterface
-from ..interface import RichJsonInterface
+from ..interface import Interface
 from ..response import check_response
 from ._aux_task_caching import FractalCacheError
 from ._aux_task_caching import get_task_id_from_cache
@@ -16,10 +14,9 @@
 
 def post_workflow(
     client: AuthClient, *, name: str, project_id: int, batch: bool = False
-) -> BaseInterface:
+) -> Interface:
     workflow = dict(
         name=name,
-        project_id=project_id,
     )
     res = client.post(
         f"{settings.BASE_URL}/project/{project_id}/workflow/",
@@ -27,38 +24,38 @@ def post_workflow(
     )
     workflow = check_response(res, expected_status_code=201)
     if batch:
-        return PrintInterface(retcode=0, data=workflow["id"])
+        return Interface(retcode=0, data=workflow["id"])
     else:
-        return RichJsonInterface(retcode=0, data=workflow)
+        return Interface(retcode=0, data=workflow)
 
 
 def get_workflow_list(
     client: AuthClient, *, project_id: int, batch: bool = False
-) -> RichJsonInterface:
+) -> Interface:
 
     res = client.get(f"{settings.BASE_URL}/project/{project_id}/workflow/")
     workflow_list = check_response(res, expected_status_code=200)
-    return RichJsonInterface(retcode=0, data=workflow_list)
+    return Interface(retcode=0, data=workflow_list)
 
 
 def delete_workflow(
     client: AuthClient, *, project_id: int, workflow_id: int
-) -> BaseInterface:
+) -> Interface:
     res = client.delete(
         f"{settings.BASE_URL}/project/{project_id}/workflow/{workflow_id}/"
     )
     check_response(res, expected_status_code=204)
-    return PrintInterface(retcode=0, data="")
+    return Interface(retcode=0, data="")
 
 
 def get_workflow(
     client: AuthClient, *, project_id: int, workflow_id: int
-) -> RichJsonInterface:
+) -> Interface:
     res = client.get(
         f"{settings.BASE_URL}/project/{project_id}/workflow/{workflow_id}/"
     )
     workflow = check_response(res, expected_status_code=200)
-    return RichJsonInterface(retcode=0, data=workflow)
+    return Interface(retcode=0, data=workflow)
 
 
 def post_workflowtask(
@@ -66,14 +63,17 @@ def post_workflowtask(
     *,
     project_id: int,
     workflow_id: int,
+    input_filters: Optional[str] = None,
+    args_non_parallel: Optional[str] = None,
+    args_parallel: Optional[str] = None,
+    meta_non_parallel: Optional[str] = None,
+    meta_parallel: Optional[str] = None,
     task_id: Optional[int] = None,
     task_name: Optional[str] = None,
     task_version: Optional[str] = None,
     batch: bool = False,
     order: Optional[int] = None,
-    args_file: Optional[str] = None,
-    meta_file: Optional[str] = None,
-) -> RichJsonInterface:
+) -> Interface:
 
     if task_id:
         if task_version:
@@ -95,14 +95,31 @@ def post_workflowtask(
         workflow_task = dict()
     else:
         workflow_task = dict(order=order)
-    if args_file:
-        with Path(args_file).open("r") as f:
-            args = json.load(f)
-            workflow_task["args"] = args
-    if meta_file:
-        with Path(meta_file).open("r") as f:
-            meta = json.load(f)
-            workflow_task["meta"] = meta
+
+    if input_filters:
+        with Path(input_filters).open("r") as f:
+            i_filters = json.load(f)
+            workflow_task["input_filters"] = i_filters
+
+    if args_non_parallel:
+        with Path(args_non_parallel).open("r") as f:
+            a_n_p = json.load(f)
+            workflow_task["args_non_parallel"] = a_n_p
+
+    if args_parallel:
+        with Path(args_parallel).open("r") as f:
+            a_p = json.load(f)
+            workflow_task["args_parallel"] = a_p
+
+    if meta_non_parallel:
+        with Path(meta_non_parallel).open("r") as f:
+            m_n_p = json.load(f)
+            workflow_task["meta_non_parallel"] = m_n_p
+
+    if meta_parallel:
+        with Path(meta_parallel).open("r") as f:
+            m_p = json.load(f)
+            workflow_task["meta_parallel"] = m_p
 
     res = client.post(
         (
@@ -115,9 +132,9 @@ def post_workflowtask(
     workflow_task = check_response(res, expected_status_code=201)
 
     if batch:
-        return PrintInterface(retcode=0, data=str(workflow_task["id"]))
+        return Interface(retcode=0, data=str(workflow_task["id"]))
     else:
-        return RichJsonInterface(retcode=0, data=workflow_task)
+        return Interface(retcode=0, data=workflow_task)
 
 
 def patch_workflowtask(
@@ -126,28 +143,38 @@ def patch_workflowtask(
     project_id: int,
     workflow_id: int,
     workflow_task_id: int,
-    args_file: Optional[str] = None,
-    meta_file: Optional[str] = None,
-) -> RichJsonInterface:
-
-    # Check that at least one of args_file or meta_file was given (note: it
-    # would be reasonable to check it in the parser, but we are not aware of a
-    # method within argparse).
-    if not (args_file or meta_file):
-        raise ValueError(
-            "At least one of {args_file,meta_file} arguments is required"
-        )
-
-    # Combine args/meta payload
+    input_filters: Optional[str] = None,
+    args_non_parallel: Optional[str] = None,
+    args_parallel: Optional[str] = None,
+    meta_non_parallel: Optional[str] = None,
+    meta_parallel: Optional[str] = None,
+) -> Interface:
+
     payload = {}
-    if args_file:
-        with Path(args_file).open("r") as f:
-            args = json.load(f)
-            payload["args"] = args
-    if meta_file:
-        with Path(meta_file).open("r") as f:
-            meta = json.load(f)
-            payload["meta"] = meta
+    if input_filters:
+        with Path(input_filters).open("r") as f:
+            input_filters = json.load(f)
+            payload["input_filters"] = input_filters
+
+    if args_non_parallel:
+        with Path(args_non_parallel).open("r") as f:
+            a_n_p = json.load(f)
+            payload["args_non_parallel"] = a_n_p
+
+    if args_parallel:
+        with Path(args_parallel).open("r") as f:
+            a_p = json.load(f)
+            payload["args_parallel"] = a_p
+
+    if meta_non_parallel:
+        with Path(meta_non_parallel).open("r") as f:
+            m_n_p = json.load(f)
+            payload["meta_non_parallel"] = m_n_p
+
+    if meta_parallel:
+        with Path(meta_parallel).open("r") as f:
+            m_p = json.load(f)
+            payload["meta_parallel"] = m_p
 
     res = client.patch(
         (
@@ -158,7 +185,7 @@ def patch_workflowtask(
     )
     workflow_task = check_response(res, expected_status_code=200)
 
-    return RichJsonInterface(retcode=0, data=workflow_task)
+    return Interface(retcode=0, data=workflow_task)
 
 
 def delete_workflowtask(
@@ -167,14 +194,14 @@ def delete_workflowtask(
     project_id: int,
     workflow_id: int,
     workflow_task_id: int,
-) -> BaseInterface:
+) -> Interface:
 
     res = client.delete(
         f"{settings.BASE_URL}/project/{project_id}/"
         f"workflow/{workflow_id}/wftask/{workflow_task_id}/"
     )
     check_response(res, expected_status_code=204)
-    return PrintInterface(retcode=0, data="")
+    return Interface(retcode=0, data="")
 
 
 def patch_workflow(
@@ -183,7 +210,7 @@ def patch_workflow(
     project_id: int,
     workflow_id: int,
     new_name: str,
-) -> BaseInterface:
+) -> Interface:
 
     workflow_update = dict(name=new_name)
 
@@ -192,121 +219,12 @@ def patch_workflow(
         json=workflow_update,
     )
     new_workflow = check_response(res, expected_status_code=200)
-    return RichJsonInterface(retcode=0, data=new_workflow)
-
-
-def workflow_apply(
-    client: AuthClient,
-    *,
-    project_id: int,
-    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,
-    batch: bool = False,
-) -> BaseInterface:
-    apply_wf_create = dict(
-        workflow_id=workflow_id,
-        input_dataset_id=input_dataset_id,
-        output_dataset_id=output_dataset_id,
-    )
-    # Prepare ApplyWorkflowCreate object, without None attributes
-    if worker_init is not None:
-        apply_wf_create["worker_init"] = worker_init
-    if first_task_index is not None:
-        apply_wf_create["first_task_index"] = first_task_index
-    if last_task_index is not None:
-        apply_wf_create["last_task_index"] = last_task_index
-
-    # Prepare query parameters
-    query_parameters = (
-        f"input_dataset_id={input_dataset_id}"
-        f"&output_dataset_id={output_dataset_id}"
-    )
-
-    res = client.post(
-        (
-            f"{settings.BASE_URL}/project/{project_id}/workflow/{workflow_id}/"
-            f"apply/?{query_parameters}"
-        ),
-        json=apply_wf_create,
-    )
-    apply_wf_read = check_response(res, expected_status_code=202)
-
-    if batch:
-        return PrintInterface(retcode=0, data=apply_wf_read["id"])
-    else:
-        return RichJsonInterface(retcode=0, data=apply_wf_read)
+    return Interface(retcode=0, data=new_workflow)
 
 
-def workflow_import(
-    client: AuthClient, *, project_id: int, json_file: str, batch: bool = False
-) -> BaseInterface:
-    with Path(json_file).open("r") as f:
-        workflow = json.load(f)
+def workflow_import() -> None:
+    raise NotImplementedError
 
-    res = client.post(
-        f"{settings.BASE_URL}/project/{project_id}/workflow/import/",
-        json=workflow,
-    )
-    wf_read = check_response(res, expected_status_code=201)
-
-    warnings = [
-        workflow_task["task"]["source"]
-        for workflow_task in wf_read["task_list"]
-        if workflow_task["task"]["owner"]
-    ]
-    if warnings:
-        sources_str = ", ".join([f"'{s}'" for s in warnings])
-        logging.warning(
-            "This workflow includes custom tasks (the ones with sources: "
-            f"{sources_str}), which are not meant to be portable; "
-            "importing this workflow may not work as expected."
-        )
 
-    if batch:
-        datastr = f"{wf_read['id']}"
-        for wftask in wf_read["task_list"]:
-            datastr += f" {wftask['id']}"
-        return PrintInterface(retcode=0, data=datastr)
-    else:
-        return RichJsonInterface(retcode=0, data=wf_read)
-
-
-def workflow_export(
-    client: AuthClient,
-    *,
-    project_id: int,
-    workflow_id: int,
-    json_file: str,
-) -> BaseInterface:
-    res = client.get(
-        (
-            f"{settings.BASE_URL}/project/{project_id}/"
-            f"workflow/{workflow_id}/export/"
-        ),
-    )
-    workflow = check_response(res, expected_status_code=200)
-
-    warnings = [
-        workflow_task["task"]["source"]
-        for workflow_task in workflow["task_list"]
-        if not workflow_task["task"]["source"].startswith(
-            ("pip_local:", "pip_remote:")
-        )
-    ]
-    if warnings:
-        sources_str = ", ".join([f"'{s}'" for s in warnings])
-        logging.warning(
-            "This workflow includes custom tasks (the ones with sources: "
-            f"{sources_str}), which are not meant to be portable; "
-            "re-importing this workflow may not work as expected."
-        )
-
-    with Path(json_file).open("w") as f:
-        json.dump(workflow, f, indent=2)
-    return PrintInterface(
-        retcode=0, data=f"Workflow {workflow_id} exported at {json_file}"
-    )
+def workflow_export() -> None:
+    raise NotImplementedError
diff --git a/fractal_client/config.py b/fractal_client/config.py
index 5886ea2fb..fd5de3207 100644
--- a/fractal_client/config.py
+++ b/fractal_client/config.py
@@ -35,7 +35,7 @@ def __init__(self):
 
     @property
     def BASE_URL(self) -> str:
-        return f"{self.FRACTAL_SERVER}/api/v1"
+        return f"{self.FRACTAL_SERVER}/api/v2"
 
 
 settings = Settings()
diff --git a/fractal_client/interface.py b/fractal_client/interface.py
index 8f13b3ba2..58072a25e 100644
--- a/fractal_client/interface.py
+++ b/fractal_client/interface.py
@@ -1,65 +1,14 @@
+import json
 from typing import Any
-from typing import Dict
-from typing import List
-from typing import Optional
-from typing import Sequence
-from typing import Union
 
-from rich import print_json
-from rich.console import Console
 
-
-class BaseInterface:
-    def __init__(self, retcode: int, data=None):
+class Interface:
+    def __init__(self, retcode: int, data: Any) -> None:
         self.retcode = retcode
         self.data = data
 
     def show(self, *args, **kwargs):
-        raise NotImplementedError("Implement in subclasses")
-
-    def __repr__(self):
-        return f"<Interface (retcode={self.retcode})>"
-
-
-class PrintInterface(BaseInterface):
-    """
-    Basic interface that simply prints a string to screen
-    """
-
-    def __init__(self, retcode: int, data: Any):
-        super().__init__(retcode, data)
-
-    def show(self, *args, **kwargs):
-        print(str(self.data))
-
-
-class RichJsonInterface(BaseInterface):
-    """
-    Output json using rich.print_json
-    """
-
-    def __init__(
-        self,
-        retcode: int,
-        data: Union[Dict[str, Any], List[Dict[str, Any]]],
-        extra_lines: Optional[Sequence[str]] = None,
-    ):
-        super().__init__(retcode, data)
-        if extra_lines is None:
-            self.extra_lines = None
+        if isinstance(self.data, (dict, list)):
+            print(json.dumps(self.data, indent=2, sort_keys=True))
         else:
-            self.extra_lines = "\n".join(extra_lines)
-
-    def show(self, *args, **kwargs):
-        print_json(data=self.data)
-        if self.extra_lines:
-            print(self.extra_lines)
-
-
-class RichConsoleInterface(BaseInterface):
-    def __init__(self, retcode: int, data: Any):
-        super().__init__(retcode, data)
-
-    def show(self, *args, **kwargs):
-        console = Console()
-        console.print(self.data)
+            print(self.data)
diff --git a/fractal_client/parser.py b/fractal_client/parser.py
index ad9141a11..b25f25dae 100644
--- a/fractal_client/parser.py
+++ b/fractal_client/parser.py
@@ -100,25 +100,21 @@
     allow_abbrev=False,
 )
 project_add_dataset_parser.add_argument(
-    "project_id", type=int, help="ID of project to add the new dataset to."
-)
-project_add_dataset_parser.add_argument(
-    "dataset_name", help="Name of new dataset."
+    "project_id",
+    type=int,
+    help="ID of project to add the new dataset to.",
 )
 project_add_dataset_parser.add_argument(
-    "--metadata",
-    help="Path to file containing dataset metadata in JSON format.",
+    "dataset_name",
+    help="Name of new dataset.",
 )
 project_add_dataset_parser.add_argument(
-    "--type",
-    help="Dataset type.",
+    "zarr_dir",
+    help="Path to zarr dir.",
 )
 project_add_dataset_parser.add_argument(
-    "--make-read-only",
-    action="store_true",
-    default=False,
-    required=False,
-    help="Make dataset read-only (not required).",
+    "--filters",
+    help="Path to JSON file with filters.",
 )
 
 
@@ -134,21 +130,6 @@
 project_edit_parser.add_argument(
     "--new-name", help="New project name.", required=False
 )
-project_edit_parser_read_only = (
-    project_edit_parser.add_mutually_exclusive_group()
-)
-project_edit_parser_read_only.add_argument(
-    "--make-read-only",
-    help="Set the read-only flag for this project.",
-    action="store_true",
-    required=False,
-)
-project_edit_parser_read_only.add_argument(
-    "--remove-read-only",
-    help="Remove the read-only flag for this project.",
-    action="store_true",
-    required=False,
-)
 
 
 # DATASET GROUP
@@ -161,37 +142,6 @@
     title="Valid sub-commands", dest="subcmd", required=True
 )
 
-# dataset add-resource
-dataset_add_resource_parser = dataset_subparsers.add_parser(
-    "add-resource",
-    description="Add resource to existing dataset.",
-    allow_abbrev=False,
-)
-dataset_add_resource_parser.add_argument(
-    "project_id", type=int, help="Project ID."
-)
-dataset_add_resource_parser.add_argument(
-    "dataset_id", type=int, help="Dataset ID."
-)
-dataset_add_resource_parser.add_argument(
-    "path", help="Absolute path to resource."
-)
-
-# dataset rm-resource
-dataset_rm_resource_parser = dataset_subparsers.add_parser(
-    "rm-resource",
-    description="Remove resource to existing dataset.",
-    allow_abbrev=False,
-)
-dataset_rm_resource_parser.add_argument(
-    "project_id", type=int, help="Project ID."
-)
-dataset_rm_resource_parser.add_argument(
-    "dataset_id", type=int, help="Dataset ID."
-)
-dataset_rm_resource_parser.add_argument(
-    "resource_id", type=int, help="Resource ID."
-)
 
 # dataset edit
 dataset_edit_parser = dataset_subparsers.add_parser(
@@ -203,24 +153,11 @@
 dataset_edit_parser.add_argument("project_id", type=int, help="Project ID.")
 dataset_edit_parser.add_argument("dataset_id", type=int, help="Dataset ID.")
 dataset_edit_parser.add_argument("--new-name", help="New name of dataset.")
-dataset_edit_parser.add_argument("--new-type", help="Dataset type.")
 dataset_edit_parser.add_argument(
-    "--meta-file",
+    "--filters",
     help="Path to JSON file with new metadata to replace the current ones.",
 )
-dataset_edit_parser_read = dataset_edit_parser.add_mutually_exclusive_group()
-dataset_edit_parser_read.add_argument(
-    "--make-read-only",
-    help="Set read-only flag of dataset.",
-    action="store_true",
-    required=False,
-)
-dataset_edit_parser_read.add_argument(
-    "--remove-read-only",
-    help="Remove read-only flag of dataset.",
-    action="store_true",
-    required=False,
-)
+
 
 # dataset show
 dataset_show_parser = dataset_subparsers.add_parser(
@@ -242,26 +179,6 @@
 dataset_delete_parser.add_argument("project_id", type=int, help="Project ID.")
 dataset_delete_parser.add_argument("dataset_id", type=int, help="Dataset ID.")
 
-# dataset history
-dataset_history_parser = dataset_subparsers.add_parser(
-    "history",
-    description="Export dataset history as a reproducible worfklow.",
-    argument_default=ap.SUPPRESS,
-    allow_abbrev=False,
-)
-dataset_history_parser.add_argument("project_id", type=int, help="Project ID.")
-dataset_history_parser.add_argument("dataset_id", type=int, help="Dataset ID.")
-
-# dataset status
-dataset_status_parser = dataset_subparsers.add_parser(
-    "status",
-    description="Extract status of WorkflowTasks associated with dataset.",
-    argument_default=ap.SUPPRESS,
-    allow_abbrev=False,
-)
-dataset_status_parser.add_argument("project_id", type=int, help="Project ID.")
-dataset_status_parser.add_argument("dataset_id", type=int, help="Dataset ID.")
-
 # TASK GROUP
 task_parser = subparsers_main.add_parser(
     "task",
@@ -336,142 +253,6 @@
     action="store_true",
     help="Also include task-collection logs.",
 )
-task_check_collection_parser.add_argument(
-    "--do-not-separate-logs",
-    dest="do_not_separate_logs",
-    help=(
-        "Show the task-collection logs in the main output, "
-        "instead of a separate field."
-    ),
-    action="store_true",
-    required=False,
-)
-
-
-# task new
-task_new_parser = task_subparsers.add_parser(
-    "new",
-    description="Create new task.",
-    argument_default=ap.SUPPRESS,
-    allow_abbrev=False,
-)
-task_new_parser.add_argument(
-    "name", help="A human readable name for the task."
-)
-task_new_parser.add_argument(
-    "command", help="The command that executes the task."
-)
-task_new_parser.add_argument("source", help="TBD")
-task_new_parser.add_argument(
-    "--input-type",
-    help="The type of data the task expects as input.",
-    default="Any",
-)
-task_new_parser.add_argument(
-    "--output-type",
-    help="The type of data the task expects as output.",
-    default="Any",
-)
-task_new_parser.add_argument(
-    "--version",
-    help="Task version.",
-)
-task_new_parser.add_argument(
-    "--meta-file",
-    help="Path to JSON file with additional parameters useful for execution.",
-)
-task_new_parser.add_argument(
-    "--args-schema",
-    help="Path to file containing JSON Schema for task arguments.",
-)
-task_new_parser.add_argument(
-    "--args-schema-version",
-    help=(
-        "Label encoding how the task-arguments JSON Schema was generated "
-        "(e.g. `pydantic_v1`)."
-    ),
-)
-
-# task edit
-task_edit_parser = task_subparsers.add_parser(
-    "edit",
-    description="Edit task.",
-    argument_default=ap.SUPPRESS,
-    allow_abbrev=False,
-)
-
-task_edit_id_or_name_group = task_edit_parser.add_mutually_exclusive_group(
-    required=True
-)
-task_edit_id_or_name_group.add_argument(
-    "--id", help="ID of the task to edit.", type=int
-)
-task_edit_id_or_name_group.add_argument(
-    "--name", help="Name of the task to edit."
-)
-
-task_edit_parser.add_argument(
-    "--version",
-    help=(
-        "Version of the task to edit "
-        "(only accepted in combination with `--name`)."
-    ),
-)
-task_edit_parser.add_argument("--new-name", help="New task name.")
-task_edit_parser.add_argument("--new-command", help="New task command.")
-task_edit_parser.add_argument(
-    "--new-input-type",
-    help="New input type.",
-)
-task_edit_parser.add_argument(
-    "--new-output-type",
-    help="New output type.",
-)
-task_edit_parser.add_argument(
-    "--meta-file",
-    help=(
-        "Path to JSON serialised file containing updates to the current "
-        "`meta` dictionary."
-    ),
-)
-task_edit_parser.add_argument(
-    "--new-version",
-    help="New version.",
-)
-task_edit_parser.add_argument(
-    "--new-args-schema",
-    help="Path to file containing the new JSON Schema for task arguments.",
-)
-task_edit_parser.add_argument(
-    "--new-args-schema-version",
-    help=(
-        "New label encoding how the task-arguments JSON Schema was generated."
-    ),
-)
-
-# task delete
-task_delete_parser = task_subparsers.add_parser(
-    "delete",
-    description="Delete task.",
-    argument_default=ap.SUPPRESS,
-    allow_abbrev=False,
-)
-task_delete_id_or_name_group = task_delete_parser.add_mutually_exclusive_group(
-    required=True
-)
-task_delete_id_or_name_group.add_argument(
-    "--id", help="ID of the task to delete.", type=int
-)
-task_delete_id_or_name_group.add_argument(
-    "--name", help="Name of the task to delete."
-)
-task_delete_parser.add_argument(
-    "--version",
-    help=(
-        "Version of the task to delete "
-        "(only accepted in combination with `--name`)."
-    ),
-)
 
 
 # WORKFLOW GROUP
@@ -565,7 +346,6 @@
 workflow_add_task_id_or_name_group.add_argument(
     "--task-name", help="Name of the task to add."
 )
-
 workflow_add_task_parser.add_argument(
     "--task-version",
     help=(
@@ -576,21 +356,29 @@
 workflow_add_task_parser.add_argument(
     "--order", help="Order of this task within the workflow's task list."
 )
+
 workflow_add_task_parser.add_argument(
-    "--args-file",
-    help=(
-        "Path to json serialised file containing the arguments "
-        "overrides of the task."
-    ),
+    "--args-non-parallel", help="Args for non parallel tasks"
 )
+
 workflow_add_task_parser.add_argument(
-    "--meta-file",
-    help=(
-        "Path to json serialised file containing the meta "
-        "overrides of the task."
-    ),
+    "--args-parallel", help="Args for parallel tasks"
+)
+
+workflow_add_task_parser.add_argument(
+    "--meta-non-parallel", help="Metadata file fornon parallel tasks"
+)
+
+workflow_add_task_parser.add_argument(
+    "--meta-parallel", help="Metadata file for parallel tasks"
+)
+
+workflow_add_task_parser.add_argument(
+    "--input-filters",
+    help="Path to json file with filters.",
 )
 
+
 # workflow edit task
 workflow_edit_task_parser = workflow_subparsers.add_parser(
     "edit-task",
@@ -611,20 +399,29 @@
     help="Workflow task ID, the ID of a task inside the list of tasks.",
 )
 workflow_edit_task_parser.add_argument(
-    "--args-file",
+    "--input-filters",
     help=(
         "Path to json serialised file containing the arguments "
         "overrides of the task."
     ),
 )
 workflow_edit_task_parser.add_argument(
-    "--meta-file",
-    help=(
-        "Path to json serialised file containing the meta "
-        "overrides of the task."
-    ),
+    "--args-non-parallel", help="Args for non parallel tasks"
+)
+
+workflow_edit_task_parser.add_argument(
+    "--args-parallel", help="Args for parallel tasks"
+)
+
+workflow_edit_task_parser.add_argument(
+    "--meta-non-parallel", help="Metadata file fornon parallel tasks"
+)
+
+workflow_edit_task_parser.add_argument(
+    "--meta-parallel", help="Metadata file for parallel tasks"
 )
 
+
 # workflow remove task
 workflow_remove_task_parser = workflow_subparsers.add_parser(
     "rm-task",
@@ -645,43 +442,6 @@
     help="Workflow task ID (the ID of a task inside the list of tasks).",
 )
 
-# workflow apply
-workflow_apply_parser = workflow_subparsers.add_parser(
-    "apply",
-    description="Apply workflow to dataset.",
-    argument_default=ap.SUPPRESS,
-    allow_abbrev=False,
-)
-
-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"
-        " (starting from 0)."
-    ),
-    required=False,
-)
-workflow_apply_parser.add_argument(
-    "--end",
-    dest="last_task_index",
-    type=int,
-    help=(
-        "Positional index of the last task to be executed"
-        " (starting from 0)."
-    ),
-    required=False,
-)
-workflow_apply_parser.add_argument(
-    "-w",
-    "--worker-init",
-    help="Command to be run before starting a worker.",
-)
 
 # workflow import
 workflow_import_parser = workflow_subparsers.add_parser(
@@ -754,13 +514,6 @@
 )
 job_show_parser.add_argument("project_id", type=int, help="Project ID.")
 job_show_parser.add_argument("job_id", type=int, help="Job ID.")
-job_show_parser.add_argument(
-    "--do-not-separate-logs",
-    dest="do_not_separate_logs",
-    help="Show the job logs in the main output, instead of a separate field.",
-    action="store_true",
-    required=False,
-)
 
 # job download-logs
 job_download_logs_parser = job_subparsers.add_parser(
@@ -789,6 +542,44 @@
 job_stop_parser.add_argument("job_id", type=int, help="Job ID.")
 
 
+# job submit
+job_submit_parser = job_subparsers.add_parser(
+    "submit",
+    description="Submit a job.",
+    argument_default=ap.SUPPRESS,
+    allow_abbrev=False,
+)
+
+job_submit_parser.add_argument("project_id", type=int)
+job_submit_parser.add_argument("workflow_id", type=int)
+job_submit_parser.add_argument("dataset_id", type=int)
+job_submit_parser.add_argument(
+    "--start",
+    dest="first_task_index",
+    type=int,
+    help=(
+        "Positional index of the first task to be executed"
+        " (starting from 0)."
+    ),
+    required=False,
+)
+job_submit_parser.add_argument(
+    "--end",
+    dest="last_task_index",
+    type=int,
+    help=(
+        "Positional index of the last task to be executed"
+        " (starting from 0)."
+    ),
+    required=False,
+)
+job_submit_parser.add_argument(
+    "-w",
+    "--worker-init",
+    help="Command to be run before starting a worker.",
+)
+
+
 # VERSION GROUP
 version_parser = subparsers_main.add_parser(
     "version",
diff --git a/poetry.lock b/poetry.lock
index 5dbf330fe..cf1ee405b 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 1.7.1 and should not be changed by hand.
+# This file is automatically @generated by Poetry 1.8.2 and should not be changed by hand.
 
 [[package]]
 name = "aiosqlite"
@@ -365,20 +365,6 @@ files = [
     {file = "colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44"},
 ]
 
-[[package]]
-name = "commonmark"
-version = "0.9.1"
-description = "Python parser for the CommonMark Markdown spec"
-optional = false
-python-versions = "*"
-files = [
-    {file = "commonmark-0.9.1-py2.py3-none-any.whl", hash = "sha256:da2f38c92590f83de410ba1a3cbceafbc74fee9def35f9251ba9a971d6d66fd9"},
-    {file = "commonmark-0.9.1.tar.gz", hash = "sha256:452f9dc859be7f06631ddcb328b6919c67984aca654e5fefb3914d54691aed60"},
-]
-
-[package.extras]
-test = ["flake8 (==3.7.8)", "hypothesis (==3.55.3)"]
-
 [[package]]
 name = "coverage"
 version = "6.5.0"
@@ -650,7 +636,7 @@ typing = ["typing-extensions (>=4.8)"]
 
 [[package]]
 name = "fractal-server"
-version = "2.0.0a0"
+version = "2.0.0a2"
 description = "Server component of the Fractal analytics platform"
 optional = false
 python-versions = "^3.9"
@@ -680,7 +666,7 @@ postgres = ["asyncpg (>=0.29.0,<0.30.0)", "psycopg2 (>=2.9.5,<3.0.0)"]
 type = "git"
 url = "https://github.com/fractal-analytics-platform/fractal-server.git"
 reference = "main"
-resolved_reference = "8f6267f9d4e30c5414edabfb115bdaae0485f9f0"
+resolved_reference = "e6d9833a3fb487b305d96f11834bd3a0dd035b38"
 
 [[package]]
 name = "ghp-import"
@@ -841,17 +827,17 @@ socks = ["socksio (==1.*)"]
 
 [[package]]
 name = "httpx-oauth"
-version = "0.13.2"
+version = "0.13.3"
 description = "Async OAuth client using HTTPX"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "httpx_oauth-0.13.2-py3-none-any.whl", hash = "sha256:9034cbb586590e59006d8a7c8c503c61e530f6dba317f998d4ccb43d78ac4351"},
-    {file = "httpx_oauth-0.13.2.tar.gz", hash = "sha256:4dccae219170d76e745cb6ad76b6d927effad59c25b9098b8b527ca1039522fc"},
+    {file = "httpx_oauth-0.13.3-py3-none-any.whl", hash = "sha256:83da5e5ba67dabced02900b7adf1a805d3c77c96d74797772fad958c95fc6be3"},
+    {file = "httpx_oauth-0.13.3.tar.gz", hash = "sha256:566d7147a2cf861fba4fa42ab065e766622e2f142bdc29ab6a9a6b630eb8d1ff"},
 ]
 
 [package.dependencies]
-httpx = ">=0.18,<0.27"
+httpx = ">=0.18,<0.28"
 
 [[package]]
 name = "identify"
@@ -1370,47 +1356,47 @@ files = [
 
 [[package]]
 name = "pydantic"
-version = "1.10.14"
+version = "1.10.15"
 description = "Data validation and settings management using python type hints"
 optional = false
 python-versions = ">=3.7"
 files = [
-    {file = "pydantic-1.10.14-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:7f4fcec873f90537c382840f330b90f4715eebc2bc9925f04cb92de593eae054"},
-    {file = "pydantic-1.10.14-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:8e3a76f571970fcd3c43ad982daf936ae39b3e90b8a2e96c04113a369869dc87"},
-    {file = "pydantic-1.10.14-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:82d886bd3c3fbeaa963692ef6b643159ccb4b4cefaf7ff1617720cbead04fd1d"},
-    {file = "pydantic-1.10.14-cp310-cp310-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:798a3d05ee3b71967844a1164fd5bdb8c22c6d674f26274e78b9f29d81770c4e"},
-    {file = "pydantic-1.10.14-cp310-cp310-musllinux_1_1_i686.whl", hash = "sha256:23d47a4b57a38e8652bcab15a658fdb13c785b9ce217cc3a729504ab4e1d6bc9"},
-    {file = "pydantic-1.10.14-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:f9f674b5c3bebc2eba401de64f29948ae1e646ba2735f884d1594c5f675d6f2a"},
-    {file = "pydantic-1.10.14-cp310-cp310-win_amd64.whl", hash = "sha256:24a7679fab2e0eeedb5a8924fc4a694b3bcaac7d305aeeac72dd7d4e05ecbebf"},
-    {file = "pydantic-1.10.14-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:9d578ac4bf7fdf10ce14caba6f734c178379bd35c486c6deb6f49006e1ba78a7"},
-    {file = "pydantic-1.10.14-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:fa7790e94c60f809c95602a26d906eba01a0abee9cc24150e4ce2189352deb1b"},
-    {file = "pydantic-1.10.14-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:aad4e10efa5474ed1a611b6d7f0d130f4aafadceb73c11d9e72823e8f508e663"},
-    {file = "pydantic-1.10.14-cp311-cp311-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:1245f4f61f467cb3dfeced2b119afef3db386aec3d24a22a1de08c65038b255f"},
-    {file = "pydantic-1.10.14-cp311-cp311-musllinux_1_1_i686.whl", hash = "sha256:21efacc678a11114c765eb52ec0db62edffa89e9a562a94cbf8fa10b5db5c046"},
-    {file = "pydantic-1.10.14-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:412ab4a3f6dbd2bf18aefa9f79c7cca23744846b31f1d6555c2ee2b05a2e14ca"},
-    {file = "pydantic-1.10.14-cp311-cp311-win_amd64.whl", hash = "sha256:e897c9f35281f7889873a3e6d6b69aa1447ceb024e8495a5f0d02ecd17742a7f"},
-    {file = "pydantic-1.10.14-cp37-cp37m-macosx_10_9_x86_64.whl", hash = "sha256:d604be0f0b44d473e54fdcb12302495fe0467c56509a2f80483476f3ba92b33c"},
-    {file = "pydantic-1.10.14-cp37-cp37m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:a42c7d17706911199798d4c464b352e640cab4351efe69c2267823d619a937e5"},
-    {file = "pydantic-1.10.14-cp37-cp37m-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:596f12a1085e38dbda5cbb874d0973303e34227b400b6414782bf205cc14940c"},
-    {file = "pydantic-1.10.14-cp37-cp37m-musllinux_1_1_i686.whl", hash = "sha256:bfb113860e9288d0886e3b9e49d9cf4a9d48b441f52ded7d96db7819028514cc"},
-    {file = "pydantic-1.10.14-cp37-cp37m-musllinux_1_1_x86_64.whl", hash = "sha256:bc3ed06ab13660b565eed80887fcfbc0070f0aa0691fbb351657041d3e874efe"},
-    {file = "pydantic-1.10.14-cp37-cp37m-win_amd64.whl", hash = "sha256:ad8c2bc677ae5f6dbd3cf92f2c7dc613507eafe8f71719727cbc0a7dec9a8c01"},
-    {file = "pydantic-1.10.14-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:c37c28449752bb1f47975d22ef2882d70513c546f8f37201e0fec3a97b816eee"},
-    {file = "pydantic-1.10.14-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:49a46a0994dd551ec051986806122767cf144b9702e31d47f6d493c336462597"},
-    {file = "pydantic-1.10.14-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:53e3819bd20a42470d6dd0fe7fc1c121c92247bca104ce608e609b59bc7a77ee"},
-    {file = "pydantic-1.10.14-cp38-cp38-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:0fbb503bbbbab0c588ed3cd21975a1d0d4163b87e360fec17a792f7d8c4ff29f"},
-    {file = "pydantic-1.10.14-cp38-cp38-musllinux_1_1_i686.whl", hash = "sha256:336709883c15c050b9c55a63d6c7ff09be883dbc17805d2b063395dd9d9d0022"},
-    {file = "pydantic-1.10.14-cp38-cp38-musllinux_1_1_x86_64.whl", hash = "sha256:4ae57b4d8e3312d486e2498d42aed3ece7b51848336964e43abbf9671584e67f"},
-    {file = "pydantic-1.10.14-cp38-cp38-win_amd64.whl", hash = "sha256:dba49d52500c35cfec0b28aa8b3ea5c37c9df183ffc7210b10ff2a415c125c4a"},
-    {file = "pydantic-1.10.14-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:c66609e138c31cba607d8e2a7b6a5dc38979a06c900815495b2d90ce6ded35b4"},
-    {file = "pydantic-1.10.14-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:d986e115e0b39604b9eee3507987368ff8148222da213cd38c359f6f57b3b347"},
-    {file = "pydantic-1.10.14-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:646b2b12df4295b4c3148850c85bff29ef6d0d9621a8d091e98094871a62e5c7"},
-    {file = "pydantic-1.10.14-cp39-cp39-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:282613a5969c47c83a8710cc8bfd1e70c9223feb76566f74683af889faadc0ea"},
-    {file = "pydantic-1.10.14-cp39-cp39-musllinux_1_1_i686.whl", hash = "sha256:466669501d08ad8eb3c4fecd991c5e793c4e0bbd62299d05111d4f827cded64f"},
-    {file = "pydantic-1.10.14-cp39-cp39-musllinux_1_1_x86_64.whl", hash = "sha256:13e86a19dca96373dcf3190fcb8797d40a6f12f154a244a8d1e8e03b8f280593"},
-    {file = "pydantic-1.10.14-cp39-cp39-win_amd64.whl", hash = "sha256:08b6ec0917c30861e3fe71a93be1648a2aa4f62f866142ba21670b24444d7fd8"},
-    {file = "pydantic-1.10.14-py3-none-any.whl", hash = "sha256:8ee853cd12ac2ddbf0ecbac1c289f95882b2d4482258048079d13be700aa114c"},
-    {file = "pydantic-1.10.14.tar.gz", hash = "sha256:46f17b832fe27de7850896f3afee50ea682220dd218f7e9c88d436788419dca6"},
+    {file = "pydantic-1.10.15-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:22ed12ee588b1df028a2aa5d66f07bf8f8b4c8579c2e96d5a9c1f96b77f3bb55"},
+    {file = "pydantic-1.10.15-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:75279d3cac98186b6ebc2597b06bcbc7244744f6b0b44a23e4ef01e5683cc0d2"},
+    {file = "pydantic-1.10.15-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:50f1666a9940d3d68683c9d96e39640f709d7a72ff8702987dab1761036206bb"},
+    {file = "pydantic-1.10.15-cp310-cp310-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:82790d4753ee5d00739d6cb5cf56bceb186d9d6ce134aca3ba7befb1eedbc2c8"},
+    {file = "pydantic-1.10.15-cp310-cp310-musllinux_1_1_i686.whl", hash = "sha256:d207d5b87f6cbefbdb1198154292faee8017d7495a54ae58db06762004500d00"},
+    {file = "pydantic-1.10.15-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:e49db944fad339b2ccb80128ffd3f8af076f9f287197a480bf1e4ca053a866f0"},
+    {file = "pydantic-1.10.15-cp310-cp310-win_amd64.whl", hash = "sha256:d3b5c4cbd0c9cb61bbbb19ce335e1f8ab87a811f6d589ed52b0254cf585d709c"},
+    {file = "pydantic-1.10.15-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:c3d5731a120752248844676bf92f25a12f6e45425e63ce22e0849297a093b5b0"},
+    {file = "pydantic-1.10.15-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:c365ad9c394f9eeffcb30a82f4246c0006417f03a7c0f8315d6211f25f7cb654"},
+    {file = "pydantic-1.10.15-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:3287e1614393119c67bd4404f46e33ae3be3ed4cd10360b48d0a4459f420c6a3"},
+    {file = "pydantic-1.10.15-cp311-cp311-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:be51dd2c8596b25fe43c0a4a59c2bee4f18d88efb8031188f9e7ddc6b469cf44"},
+    {file = "pydantic-1.10.15-cp311-cp311-musllinux_1_1_i686.whl", hash = "sha256:6a51a1dd4aa7b3f1317f65493a182d3cff708385327c1c82c81e4a9d6d65b2e4"},
+    {file = "pydantic-1.10.15-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:4e316e54b5775d1eb59187f9290aeb38acf620e10f7fd2f776d97bb788199e53"},
+    {file = "pydantic-1.10.15-cp311-cp311-win_amd64.whl", hash = "sha256:0d142fa1b8f2f0ae11ddd5e3e317dcac060b951d605fda26ca9b234b92214986"},
+    {file = "pydantic-1.10.15-cp37-cp37m-macosx_10_9_x86_64.whl", hash = "sha256:7ea210336b891f5ea334f8fc9f8f862b87acd5d4a0cbc9e3e208e7aa1775dabf"},
+    {file = "pydantic-1.10.15-cp37-cp37m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:3453685ccd7140715e05f2193d64030101eaad26076fad4e246c1cc97e1bb30d"},
+    {file = "pydantic-1.10.15-cp37-cp37m-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:9bea1f03b8d4e8e86702c918ccfd5d947ac268f0f0cc6ed71782e4b09353b26f"},
+    {file = "pydantic-1.10.15-cp37-cp37m-musllinux_1_1_i686.whl", hash = "sha256:005655cabc29081de8243126e036f2065bd7ea5b9dff95fde6d2c642d39755de"},
+    {file = "pydantic-1.10.15-cp37-cp37m-musllinux_1_1_x86_64.whl", hash = "sha256:af9850d98fc21e5bc24ea9e35dd80a29faf6462c608728a110c0a30b595e58b7"},
+    {file = "pydantic-1.10.15-cp37-cp37m-win_amd64.whl", hash = "sha256:d31ee5b14a82c9afe2bd26aaa405293d4237d0591527d9129ce36e58f19f95c1"},
+    {file = "pydantic-1.10.15-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:5e09c19df304b8123938dc3c53d3d3be6ec74b9d7d0d80f4f4b5432ae16c2022"},
+    {file = "pydantic-1.10.15-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:7ac9237cd62947db00a0d16acf2f3e00d1ae9d3bd602b9c415f93e7a9fc10528"},
+    {file = "pydantic-1.10.15-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:584f2d4c98ffec420e02305cf675857bae03c9d617fcfdc34946b1160213a948"},
+    {file = "pydantic-1.10.15-cp38-cp38-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:bbc6989fad0c030bd70a0b6f626f98a862224bc2b1e36bfc531ea2facc0a340c"},
+    {file = "pydantic-1.10.15-cp38-cp38-musllinux_1_1_i686.whl", hash = "sha256:d573082c6ef99336f2cb5b667b781d2f776d4af311574fb53d908517ba523c22"},
+    {file = "pydantic-1.10.15-cp38-cp38-musllinux_1_1_x86_64.whl", hash = "sha256:6bd7030c9abc80134087d8b6e7aa957e43d35714daa116aced57269a445b8f7b"},
+    {file = "pydantic-1.10.15-cp38-cp38-win_amd64.whl", hash = "sha256:3350f527bb04138f8aff932dc828f154847fbdc7a1a44c240fbfff1b57f49a12"},
+    {file = "pydantic-1.10.15-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:51d405b42f1b86703555797270e4970a9f9bd7953f3990142e69d1037f9d9e51"},
+    {file = "pydantic-1.10.15-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:a980a77c52723b0dc56640ced396b73a024d4b74f02bcb2d21dbbac1debbe9d0"},
+    {file = "pydantic-1.10.15-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:67f1a1fb467d3f49e1708a3f632b11c69fccb4e748a325d5a491ddc7b5d22383"},
+    {file = "pydantic-1.10.15-cp39-cp39-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:676ed48f2c5bbad835f1a8ed8a6d44c1cd5a21121116d2ac40bd1cd3619746ed"},
+    {file = "pydantic-1.10.15-cp39-cp39-musllinux_1_1_i686.whl", hash = "sha256:92229f73400b80c13afcd050687f4d7e88de9234d74b27e6728aa689abcf58cc"},
+    {file = "pydantic-1.10.15-cp39-cp39-musllinux_1_1_x86_64.whl", hash = "sha256:2746189100c646682eff0bce95efa7d2e203420d8e1c613dc0c6b4c1d9c1fde4"},
+    {file = "pydantic-1.10.15-cp39-cp39-win_amd64.whl", hash = "sha256:394f08750bd8eaad714718812e7fab615f873b3cdd0b9d84e76e51ef3b50b6b7"},
+    {file = "pydantic-1.10.15-py3-none-any.whl", hash = "sha256:28e552a060ba2740d0d2aabe35162652c1459a0b9069fe0db7f4ee0e18e74d58"},
+    {file = "pydantic-1.10.15.tar.gz", hash = "sha256:ca832e124eda231a60a041da4f013e3ff24949d94a01154b137fc2f2a43c3ffb"},
 ]
 
 [package.dependencies]
@@ -1495,21 +1481,6 @@ tomli = {version = ">=1.0.0", markers = "python_version < \"3.11\""}
 [package.extras]
 testing = ["argcomplete", "attrs (>=19.2.0)", "hypothesis (>=3.56)", "mock", "nose", "pygments (>=2.7.2)", "requests", "setuptools", "xmlschema"]
 
-[[package]]
-name = "pytest-pretty"
-version = "1.2.0"
-description = "pytest plugin for printing summary data as I want it"
-optional = false
-python-versions = ">=3.7"
-files = [
-    {file = "pytest_pretty-1.2.0-py3-none-any.whl", hash = "sha256:6f79122bf53864ae2951b6c9e94d7a06a87ef753476acd4588aeac018f062036"},
-    {file = "pytest_pretty-1.2.0.tar.gz", hash = "sha256:105a355f128e392860ad2c478ae173ff96d2f03044692f9818ff3d49205d3a60"},
-]
-
-[package.dependencies]
-pytest = ">=7"
-rich = ">=12"
-
 [[package]]
 name = "python-dateutil"
 version = "2.9.0.post0"
@@ -1577,6 +1548,7 @@ files = [
     {file = "PyYAML-6.0.1-cp311-cp311-win_amd64.whl", hash = "sha256:bf07ee2fef7014951eeb99f56f39c9bb4af143d8aa3c21b1677805985307da34"},
     {file = "PyYAML-6.0.1-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:855fb52b0dc35af121542a76b9a84f8d1cd886ea97c84703eaa6d88e37a2ad28"},
     {file = "PyYAML-6.0.1-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:40df9b996c2b73138957fe23a16a4f0ba614f4c0efce1e9406a184b6d07fa3a9"},
+    {file = "PyYAML-6.0.1-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:a08c6f0fe150303c1c6b71ebcd7213c2858041a7e01975da3a99aed1e7a378ef"},
     {file = "PyYAML-6.0.1-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6c22bec3fbe2524cde73d7ada88f6566758a8f7227bfbf93a408a9d86bcc12a0"},
     {file = "PyYAML-6.0.1-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:8d4e9c88387b0f5c7d5f281e55304de64cf7f9c0021a3525bd3b1c542da3b0e4"},
     {file = "PyYAML-6.0.1-cp312-cp312-win32.whl", hash = "sha256:d483d2cdf104e7c9fa60c544d92981f12ad66a457afae824d146093b8c294c54"},
@@ -1765,24 +1737,6 @@ idna = {version = "*", optional = true, markers = "extra == \"idna2008\""}
 [package.extras]
 idna2008 = ["idna"]
 
-[[package]]
-name = "rich"
-version = "12.6.0"
-description = "Render rich text, tables, progress bars, syntax highlighting, markdown and more to the terminal"
-optional = false
-python-versions = ">=3.6.3,<4.0.0"
-files = [
-    {file = "rich-12.6.0-py3-none-any.whl", hash = "sha256:a4eb26484f2c82589bd9a17c73d32a010b1e29d89f1604cd9bf3a2097b81bb5e"},
-    {file = "rich-12.6.0.tar.gz", hash = "sha256:ba3a3775974105c221d31141f2c116f4fd65c5ceb0698657a11e9f295ec93fd0"},
-]
-
-[package.dependencies]
-commonmark = ">=0.9.0,<0.10.0"
-pygments = ">=2.6.0,<3.0.0"
-
-[package.extras]
-jupyter = ["ipywidgets (>=7.5.1,<8.0.0)"]
-
 [[package]]
 name = "setuptools"
 version = "69.2.0"
@@ -2089,4 +2043,4 @@ testing = ["big-O", "jaraco.functools", "jaraco.itertools", "more-itertools", "p
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.9"
-content-hash = "19a458c1991e32a56389ff79d46aabd0ec2fe6d1d89415c8312b5f5620fe453e"
+content-hash = "564f755159caa6386e2794b40e5cef65bdd6f4cddf51c489d5f092994f80a09f"
diff --git a/pyproject.toml b/pyproject.toml
index 2582a6b98..82a68c962 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -25,7 +25,6 @@ exclude = [
 [tool.poetry.dependencies]
 python = "^3.9"
 python-dotenv = "^0.21.0"
-rich = "^12.5.1"
 httpx = "^0.23.0"
 PyJWT = "^2.4.0"
 packaging = "^23.1"
@@ -36,7 +35,6 @@ pre-commit = "^2.19.0"
 pytest = "^7.1.2"
 bumpver = "^2022.1118"
 coverage = {extras = ["toml"], version = "^6.5.0"}
-pytest-pretty = "^1.1.0"
 fractal-server = { git = "https://github.com/fractal-analytics-platform/fractal-server.git", branch = "main"}
 
 


Test patch:
diff --git a/tests/data/fractal_tasks_core_alpha-0.0.1a0-py3-none-any.whl b/tests/data/fractal_tasks_core_alpha-0.0.1a0-py3-none-any.whl
deleted file mode 100644
index 8c476d963..000000000
Binary files a/tests/data/fractal_tasks_core_alpha-0.0.1a0-py3-none-any.whl and /dev/null differ
diff --git a/tests/data/fractal_tasks_dummy-0.1.0-py3-none-any.whl b/tests/data/fractal_tasks_dummy-0.1.0-py3-none-any.whl
deleted file mode 100644
index b3742ae6d..000000000
Binary files a/tests/data/fractal_tasks_dummy-0.1.0-py3-none-any.whl and /dev/null differ
diff --git a/tests/data/fractal_tasks_dummy-0.2.0-py3-none-any.whl b/tests/data/fractal_tasks_dummy-0.2.0-py3-none-any.whl
deleted file mode 100644
index 36de21df2..000000000
Binary files a/tests/data/fractal_tasks_dummy-0.2.0-py3-none-any.whl and /dev/null differ
diff --git a/tests/fixtures_testserver.py b/tests/fixtures_testserver.py
index 0bbf53eb6..fdafd085a 100644
--- a/tests/fixtures_testserver.py
+++ b/tests/fixtures_testserver.py
@@ -107,18 +107,16 @@ def db(testserver):
 
 @pytest.fixture
 def task_factory(db):
-    from fractal_server.app.models.v1.task import Task
+    from fractal_server.app.models.v2.task import TaskV2
 
     def _task_factory(**task_args_override):
         task_args = dict(
             name="test_task",
-            command="cmd",
+            type="parallel",
             source="source",
-            input_type="Any",
-            output_type="Any",
         )
         task_args.update(task_args_override)
-        t = Task(**task_args)
+        t = TaskV2(**task_args)
         db.add(t)
         db.commit()
         db.refresh(t)
@@ -129,12 +127,12 @@ def _task_factory(**task_args_override):
 
 @pytest.fixture
 def project_factory(db):
-    from fractal_server.app.models.v1.project import Project
+    from fractal_server.app.models.v2.project import ProjectV2
 
     def _project_factory(user_id=None, **project_args_override):
         project_args = dict(name="name")
         project_args.update(project_args_override)
-        p = Project(**project_args)
+        p = ProjectV2(**project_args)
         if user_id:
             from fractal_server.app.security import User
 
@@ -150,12 +148,12 @@ def _project_factory(user_id=None, **project_args_override):
 
 @pytest.fixture
 def workflow_factory(db, project_factory):
-    from fractal_server.app.models.v1.workflow import Workflow
+    from fractal_server.app.models.v2.workflow import WorkflowV2
 
     def _workflow_factory(**wf_args_override):
         wf_args = dict(name="name")
         wf_args.update(wf_args_override)
-        wf = Workflow(**wf_args)
+        wf = WorkflowV2(**wf_args)
         db.add(wf)
         db.commit()
         db.refresh(wf)
@@ -166,7 +164,7 @@ def _workflow_factory(**wf_args_override):
 
 @pytest.fixture
 def job_factory(db):
-    from fractal_server.app.models.v1.job import ApplyWorkflow
+    from fractal_server.app.models.v2.job import JobV2
     from fractal_server.utils import get_timestamp
 
     def _job_factory(**job_args_override):
@@ -179,21 +177,14 @@ def _job_factory(**job_args_override):
             first_task_index=9999,
             last_task_index=9999,
             workflow_dump={},
-            input_dataset_dump=dict(
+            dataset_dump=dict(
                 id=1,
                 name="ds-in",
+                zarr_dir="/abc",
                 read_only=False,
                 project_id=1,
-                resource_list=[dict(path="/tmp", id=1, dataset_id=1)],
-                timestamp_created=str(get_timestamp()),
-            ),
-            output_dataset_dump=dict(
-                id=2,
-                name="ds-out",
-                read_only=False,
-                project_id=1,
-                resource_list=[dict(path="/tmp", id=1, dataset_id=2)],
                 timestamp_created=str(get_timestamp()),
+                filters=dict(attributes=dict(a=1), types=dict(b=True)),
             ),
             project_dump=dict(
                 id=1,
@@ -205,7 +196,7 @@ def _job_factory(**job_args_override):
             user_email="test@test.test",
         )
         job_args.update(job_args_override)
-        j = ApplyWorkflow(**job_args)
+        j = JobV2(**job_args)
         db.add(j)
         db.commit()
         db.refresh(j)
diff --git a/tests/test_dataset.py b/tests/test_dataset.py
index 5106afebb..5d65a00ea 100644
--- a/tests/test_dataset.py
+++ b/tests/test_dataset.py
@@ -1,6 +1,5 @@
 import json
 
-import pytest
 from devtools import debug
 
 
@@ -12,214 +11,78 @@ def test_create_dataset(register_user, invoke, tmp_path):
     res = invoke("project new prj0")
     project_id = res.data["id"]
 
-    METADATA = {"some": "value"}
-    TYPE = "sometype"
+    FILTERS = {"attributes": {"a": 1}, "types": {"b": True}}
 
-    file_metadata = str(tmp_path / "metadata.json")
-    with open(file_metadata, "w") as f:
-        json.dump(METADATA, f)
+    file_filters = str(tmp_path / "filters.json")
+    with open(file_filters, "w") as f:
+        json.dump(FILTERS, f)
 
     res = invoke(
         (
-            f"project add-dataset {project_id} MyDS "
-            f"--metadata {file_metadata} "
-            f"--type {TYPE}"
+            f"project add-dataset {project_id} MyDS /tmp "
+            f"--filters {file_filters}"
         )
     )
+    debug(file_filters)
     debug(res.data)
     assert res.retcode == 0
-    assert res.data["meta"] == METADATA
-    assert res.data["type"] == TYPE
+    assert res.data["filters"] == FILTERS
 
-    res = invoke(f"--batch project add-dataset {project_id} MyNewDS")
+    res = invoke(f"--batch project add-dataset {project_id} MyNewDS /tmp")
     debug(res.data)
     assert res.retcode == 0
 
 
-def test_add_resource(register_user, invoke):
-
-    # Create a project with its default dataset
-    res = invoke("project new prj0")
-    project_id = res.data["id"]
-
-    res = invoke(f"project add-dataset {project_id} test_name")
-    dataset_id = res.data["id"]
-    assert res.data["resource_list"] == []
-
-    # Add a resource
-    PATH = "/some/path"
-    res = invoke(f"dataset add-resource {project_id} {dataset_id} {PATH}")
-    res.show()
-    assert res.retcode == 0
-    assert res.data["path"] == PATH
-    assert res.data["dataset_id"] == dataset_id
-
-
-def test_add_resource_relative_path(register_user, invoke):
-    res = invoke("project new prj0")
-    project_id = res.data["id"]
-
-    res = invoke(f"project add-dataset {project_id} test_name")
-    dataset_id = res.data["id"]
-    assert res.data["resource_list"] == []
-
-    PATH = "../new/resource/path"
-    with pytest.raises(SystemExit):
-        res = invoke(f"dataset add-resource {project_id} {dataset_id} {PATH}")
-
-    PATH = "local-folder/new/resource/path"
-    with pytest.raises(SystemExit):
-        res = invoke(f"dataset add-resource {project_id} {dataset_id} {PATH}")
-
-
 def test_edit_dataset(register_user, invoke, tmp_path):
     res = invoke("project new prj0")
     project_id = res.data["id"]
 
-    res = invoke(f"project add-dataset {project_id} test_name")
+    res = invoke(f"project add-dataset {project_id} test_name /tmp")
     dataset_id = res.data["id"]
 
-    TYPE = "this_new_type"
     NAME = "this_new_name"
-    META = {"something": "else"}
-    META_FILE = str(tmp_path / "meta.json")
-    with open(META_FILE, "w") as f:
-        json.dump(META, f)
+    FILTERS = {"attributes": {"a": 1}, "types": {"b": True}}
+    FILTERS_FILE = str(tmp_path / "meta.json")
+    with open(FILTERS_FILE, "w") as f:
+        json.dump(FILTERS, f)
 
     res = invoke(f"dataset edit {project_id} {dataset_id} --new-name {NAME}")
     res.show()
     assert res.data["name"] == NAME
     assert res.retcode == 0
 
-    res = invoke(f"dataset edit {project_id} {dataset_id} --new-type {TYPE}")
-    res.show()
-    assert res.data["name"] == NAME
-    assert res.data["type"] == TYPE
-    assert res.retcode == 0
-
-    res = invoke(f"dataset edit {project_id} {dataset_id} --make-read-only")
-    res.show()
-    assert res.data["name"] == NAME
-    assert res.data["type"] == TYPE
-    assert res.data["read_only"]
-    assert res.retcode == 0
-
-    res = invoke(f"dataset edit {project_id} {dataset_id} --remove-read-only")
-    res.show()
-    assert res.data["name"] == NAME
-    assert res.data["type"] == TYPE
-    assert not res.data["read_only"]
-    assert res.retcode == 0
-
     res = invoke(
-        f"dataset edit {project_id} {dataset_id} --meta-file {META_FILE}"
+        f"dataset edit {project_id} {dataset_id} --filters {FILTERS_FILE}"
     )
     res.show()
     assert res.data["name"] == NAME
-    assert res.data["type"] == TYPE
-    assert res.data["meta"] == META
-    assert not res.data["read_only"]
+    assert res.data["filters"]["types"] == FILTERS["types"]
     assert res.retcode == 0
 
-    with pytest.raises(SystemExit):
-        res = invoke(
-            f"dataset edit {project_id} {dataset_id} "
-            "--make-read-only --remove-read-only"
-        )
-
 
 def test_delete_dataset(register_user, invoke):
     # Create a project with its default dataset
     res = invoke("project new prj0")
     project_id = res.data["id"]
 
-    res = invoke(f"project add-dataset {project_id} test_name")
+    res = invoke(f"project add-dataset {project_id} test_name /tmp")
     dataset_id = res.data["id"]
 
     # Delete dataset
     res = invoke(f"dataset delete {project_id} {dataset_id}")
-
+    debug(res.data)
     # Check that dataset show fails
-    with pytest.raises(SystemExit):
-        res = invoke(f"dataset show {project_id} {dataset_id}")
-
-
-def test_show_dataset(register_user, invoke):
-    # Create a project with its default dataset
-    res = invoke("project new prj0")
-    project_id = res.data["id"]
-    res = invoke(f"project add-dataset {project_id} test_name")
-    dataset_id = res.data["id"]
-
     res = invoke(f"dataset show {project_id} {dataset_id}")
-    res.show()
-    assert res.retcode == 0
+    assert res.data["detail"] == "Dataset not found"
 
 
-def test_delete_resource(register_user, invoke):
+def test_show_dataset(register_user, invoke):
     # Create a project with its default dataset
     res = invoke("project new prj0")
     project_id = res.data["id"]
-    res = invoke(f"project add-dataset {project_id} test_name")
+    res = invoke(f"project add-dataset {project_id} test_name /tmp")
     dataset_id = res.data["id"]
-    assert res.data["resource_list"] == []
-
-    # Add a resource
-    PATH = "/some/path"
-    res = invoke(f"dataset add-resource {project_id} {dataset_id} {PATH}")
-    res.show()
-    assert res.retcode == 0
-    assert res.data["path"] == PATH
-    assert res.data["dataset_id"] == dataset_id
-    resource_id = res.data["id"]
-
-    # Show dataset
-    res = invoke(f"dataset show {project_id} {dataset_id}")
-    res.show()
-
-    # Remove a resource
-    res = invoke(
-        f"dataset rm-resource {project_id} {dataset_id} {resource_id}"
-    )
-    assert res.retcode == 0
 
-    # Check that the resource was removed
     res = invoke(f"dataset show {project_id} {dataset_id}")
     res.show()
-
-    # Add a new resource, and check that it has the same id as the one that was
-    # removed
-    res = invoke(
-        f"--batch dataset add-resource {project_id} {dataset_id} {PATH}"
-    )
-    assert res.retcode == 0
-    assert res.data == resource_id
-
-
-def test_dataset_history_command(register_user, invoke):
-    """
-    Only test the client interface, not the fractal-server business logic.
-    """
-    res = invoke("project new prj0")
-    project_id = res.data["id"]
-    res = invoke(f"project add-dataset {project_id} test_name")
-    dataset_id = res.data["id"]
-
-    debug(f"dataset history {project_id} {dataset_id}")
-    res = invoke(f"dataset history {project_id} {dataset_id}")
-    res.show()
-    assert res.retcode == 0
-
-
-def test_dataset_status_command(register_user, invoke):
-    """
-    Only test the client interface, not the fractal-server business logic.
-    """
-    res = invoke("project new prj0")
-    project_id = res.data["id"]
-    res = invoke(f"project add-dataset {project_id} test_name")
-    dataset_id = res.data["id"]
-
-    res = invoke(f"dataset status {project_id} {dataset_id}")
-    res.show()
     assert res.retcode == 0
diff --git a/tests/test_job.py b/tests/test_job.py
index 5451054d5..d99f6eeca 100644
--- a/tests/test_job.py
+++ b/tests/test_job.py
@@ -1,9 +1,13 @@
+import json
+import time
 from pathlib import Path
+from urllib.request import urlretrieve
 
 import pytest  # noqa F401
 from devtools import debug
 from fractal_server.utils import get_timestamp
 
+TIMEOUT = 15.0
 
 LOG = "Here are some logs"
 
@@ -54,8 +58,7 @@ def test_job_show(
     assert res.retcode == 0
     assert res.data["status"] == status
     debug(res.data)
-    assert "log" not in res.data
-    assert LOG in res.extra_lines
+    assert LOG in res.data["log"]
     res.show()
 
     # Check `job show` output with --batch
@@ -64,15 +67,6 @@ def test_job_show(
     assert res.retcode == 0
     assert res.data == status
 
-    # Check `job show` output with `--do-not-separate-logs`
-    cmd = f"job show {project_id} {job.id} --do-not-separate-logs"
-    res = invoke(cmd)
-    debug(res.data)
-    assert res.retcode == 0
-    assert res.data["status"] == status
-    debug(res.data)
-    assert res.data["log"] == LOG
-
 
 def test_job_list(
     register_user,
@@ -243,3 +237,163 @@ def test_job_stop(
     debug(caplog.text)
     assert "Stopping a job execution is not implemented" in caplog.text
     caplog.clear()
+
+
+def test_job_submit(
+    register_user, invoke, testdata_path: Path, tmp_path: Path
+):
+    """
+    GIVEN a project and a nontrivial workflow
+    WHEN the client requests to apply the workflow to the project
+    THEN the workflow is scheduled and executed, and the artifacts created
+    """
+
+    # Collect tasks
+    PACKAGE_URL = (
+        "https://github.com/fractal-analytics-platform/fractal-server/"
+        "raw/main/tests/v2/fractal_tasks_mock/dist/"
+        "fractal_tasks_mock-0.0.1-py3-none-any.whl"
+    )
+    PACKAGE_PATH = "/tmp/fractal_tasks_mock-0.0.1-py3-none-any.whl"
+    urlretrieve(PACKAGE_URL, PACKAGE_PATH)
+
+    WORKFLOW_NAME = "mywf"
+    res0 = invoke(f"task collect {PACKAGE_PATH}")
+    debug(res0)
+    venv_path = res0.data["data"]["venv_path"]
+    debug(venv_path)
+    state_id = res0.data["id"]
+
+    # Wait for task collection to end
+    starting_time = time.perf_counter()
+    while True:
+        res1 = invoke(f"task check-collection {state_id}")
+        if res1.data["data"]["status"] == "OK":
+            debug(res1.data)
+            break
+        time.sleep(1)
+        assert time.perf_counter() - starting_time < TIMEOUT
+
+    # Create a project
+    res = invoke("project new testproject")
+    assert res.retcode == 0
+    prj = res.data
+    prj_id = prj["id"]
+    res = invoke(f"project add-dataset {prj_id} test_name {tmp_path}/ds")
+    dataset_id = res.data["id"]
+
+    # Create workflow and add task twice
+    res = invoke(f"workflow new {WORKFLOW_NAME} {prj_id}")
+    workflow = res.data
+    workflow_id = workflow["id"]
+    args_file = str(tmp_path / "args_file.json")
+    with open(args_file, "w") as f:
+        json.dump({"image_dir": "/asdasd"}, f)
+    debug(workflow)
+    assert res.retcode == 0
+    for _ in [0, 1]:
+        TASK_ID = 1
+        res = invoke(
+            (
+                f"workflow add-task {prj_id} {workflow_id} "
+                f"--task-id {TASK_ID} "
+                f"--args-non-parallel {args_file}"
+            )
+        )
+        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"job submit "
+        f"{prj_id} {workflow_id} {dataset_id} "
+        f"--start {FIRST_TASK_INDEX} --end {LAST_TASK_INDEX} "
+        f'--worker-init "export SOMEVARIABLE=1"'
+    )
+    debug(cmd)
+    res = invoke(cmd)
+    job = res.data
+    debug(job)
+    assert res.retcode == 0
+    job_id = job["id"]
+    assert job["status"] == "submitted"
+
+    # Avoid immediately calling `job show` right after `workflow apply`
+    time.sleep(1)
+
+    # Check that job completed successfully
+    cmd = f"job show {prj_id} {job_id}"
+    starting_time = time.perf_counter()
+    debug(cmd)
+    while True:
+        res = invoke(cmd)
+        job = res.data
+        debug(job)
+        assert res.retcode == 0
+        if job["status"] == "done":
+            break
+        elif job["status"] == "failed":
+            raise RuntimeError(job)
+        time.sleep(1)
+        assert time.perf_counter() - starting_time < TIMEOUT
+
+    # 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:
+        json.dump({"image_dir": "/asdasd"}, f)
+    res = invoke(
+        (
+            f"workflow add-task {prj_id} {workflow_id} --task-id {TASK_ID}"
+            f" --args-non-parallel {args_file} "
+        )
+    )
+    assert res.retcode == 0
+    cmd = f"job submit " f"{prj_id} {workflow_id} {dataset_id}"
+    debug(cmd)
+    res = invoke(cmd)
+    assert res.retcode == 0
+    job_id = res.data["id"]
+
+    # Avoid immediately calling `job show` right after `workflow apply`
+    time.sleep(1)
+
+    # Verify that status is failed, and that there is a log
+    cmd = f"job show {prj_id} {job_id}"
+    starting_time = time.perf_counter()
+    while True:
+        res = invoke(cmd)
+        job = res.data
+        debug(job)
+        assert res.retcode == 0
+        if job["status"] == "failed":
+            break
+        time.sleep(1)
+        assert time.perf_counter() - starting_time < TIMEOUT
+    assert job["log"] is not None
+
+    # Prepare and submit a workflow with --batch
+    res = invoke(f"workflow new OneMoreWorkflow {prj_id}")
+    workflow_id = res.data["id"]
+    res = invoke(
+        (
+            f"workflow add-task {prj_id} {workflow_id} --task-id {TASK_ID} "
+            f"--args-non-parallel {args_file}"
+        )
+    )
+    assert res.retcode == 0
+    cmd = f"--batch job submit " f"{prj_id} {workflow_id} {dataset_id}"
+    debug(cmd)
+    res = invoke(cmd)
+    assert res.retcode == 0
+    debug(res.data)
+    assert isinstance(res.data, int)
diff --git a/tests/test_project.py b/tests/test_project.py
index 266e90a52..fb0d3efa5 100644
--- a/tests/test_project.py
+++ b/tests/test_project.py
@@ -40,53 +40,26 @@ def test_project_create_batch(register_user, invoke):
 def test_project_list(register_user, invoke):
     res = invoke("project list")
     debug(res)
-    debug(vars(res.data))
-    assert len(res.data.rows) == 0
+    assert len(res.data) == 0
 
     res.show()
 
     res = invoke("--batch project new proj0")
     project0_id = res.data
-    res = invoke(f"--batch project add-dataset {project0_id} NAME")
+    res = invoke(f"--batch project add-dataset {project0_id} NAME /tmp")
     res = invoke("--batch project new proj1")
 
     res = invoke("project list")
     debug(res)
-    debug(vars(res.data))
     res.show()
-    assert len(res.data.rows) == 2
-
-
-def test_add_dataset(register_user, invoke):
-    DATASET_NAME = "new_ds_name"
-
-    res = invoke("--batch project new proj0")
-    assert res.retcode == 0
-    debug(res.data)
-    project_id = int(res.data)
-
-    res = invoke(f"project add-dataset {project_id} {DATASET_NAME}")
-    assert res.retcode == 0
-    res.show()
-    assert res.data["name"] == DATASET_NAME
-    assert not res.data["read_only"]
-
-    res = invoke(
-        f"project add-dataset {project_id} new_{DATASET_NAME} --make-read-only"
-    )
-    assert res.retcode == 0
-    res.show()
-    assert res.data["name"] == f"new_{DATASET_NAME}"
-    assert res.data["read_only"]
+    assert len(res.data) == 2
 
 
 @pytest.mark.parametrize("new_name", ["new_name", None])
-@pytest.mark.parametrize("read_only", [True, False, None])
 def test_edit_project(
     register_user,
     invoke,
     new_name,
-    read_only,
     tmp_path,
 ):
     name = "name"
@@ -97,26 +70,13 @@ def test_edit_project(
     cmd = f"project edit {project_id}"
     if new_name:
         cmd += f" --new-name {new_name}"
-    if read_only is True:
-        cmd += " --make-read-only"
-    elif read_only is False:
-        cmd += " --remove-read-only"
 
     res = invoke(cmd)
     debug(res)
 
-    if (not new_name) and (read_only is None):
-        assert res.retcode == 1
+    assert res.retcode == 0
+    new_project = res.data
+    if new_name:
+        assert new_project["name"] == new_name
     else:
-        assert res.retcode == 0
-        new_project = res.data
-        if new_name:
-            assert new_project["name"] == new_name
-        else:
-            assert new_project["name"] == name
-        if read_only is True:
-            assert new_project["read_only"] is True
-        if read_only is False:
-            assert new_project["read_only"] is False
-        if read_only is None:
-            assert new_project["read_only"] == project["read_only"]
+        assert new_project["name"] == name
diff --git a/tests/test_task.py b/tests/test_task.py
index a1adab33a..5906bea4d 100644
--- a/tests/test_task.py
+++ b/tests/test_task.py
@@ -1,16 +1,21 @@
 import json
 import time
-from pathlib import Path
+from urllib.request import urlretrieve
 
 import pytest
 from devtools import debug
 
-from fractal_client.cmd._aux_task_caching import TASKS_CACHE_FILENAME
-from fractal_client.config import settings
-
 
 COLLECTION_TIMEOUT = 15.0
 
+PACKAGE_URL = (
+    "https://github.com/fractal-analytics-platform/fractal-server/"
+    "raw/main/tests/v2/fractal_tasks_mock/dist/"
+    "fractal_tasks_mock-0.0.1-py3-none-any.whl"
+)
+PACKAGE_PATH = "/tmp/fractal_tasks_mock-0.0.1-py3-none-any.whl"
+urlretrieve(PACKAGE_URL, PACKAGE_PATH)
+
 
 def test_task_collection_command(register_user, invoke, caplog):
     """
@@ -55,9 +60,8 @@ def test_task_collection(register_user, invoke, testdata_path):
         * the collection is initiated in the background
         * the server returns immediately
     """
-    PACKAGE = testdata_path / "fractal_tasks_dummy-0.1.0-py3-none-any.whl"
 
-    res0 = invoke(f"task collect {PACKAGE}")
+    res0 = invoke(f"task collect {PACKAGE_PATH}")
     debug(res0)
     res0.show()
 
@@ -72,6 +76,7 @@ def test_task_collection(register_user, invoke, testdata_path):
     starting_time = time.perf_counter()
     while True:
         res1 = invoke(f"task check-collection {state_id}")
+        debug(res1.data)
         assert res1.retcode == 0
         res1.show()
         time.sleep(1)
@@ -79,21 +84,14 @@ def test_task_collection(register_user, invoke, testdata_path):
             break
         assert time.perf_counter() - starting_time < COLLECTION_TIMEOUT
 
-    # Add --include-logs and --do-not-separate-logs flags
-    res2 = invoke(
-        f"task check-collection {state_id}"
-        " --include-logs"
-        " --do-not-separate-logs"
-    )
-    assert res2.retcode == 0
+    res2 = invoke(f"task check-collection {state_id}" " --include-logs")
+    debug(res2.data)
+    assert res2.retcode == 0j
     res2.show()
-    assert res2.data["data"]["log"]
     assert res2.data["data"]["status"] == "OK"
 
-    # Show again the check-collection output, without --do-not-separate-logs,
-    # for visual inspection
-    res = invoke(f"task check-collection {state_id} --include-logs")
-    res.show()
+    res = invoke("task list")
+    assert len(res.data) == 13
 
 
 def test_repeated_task_collection(register_user, invoke, testdata_path):
@@ -105,16 +103,12 @@ def test_repeated_task_collection(register_user, invoke, testdata_path):
     THEN
         * TBD..
     """
-    PACKAGE = testdata_path / "fractal_tasks_dummy-0.1.0-py3-none-any.whl"
 
-    res0 = invoke(f"task collect {PACKAGE}")
+    res0 = invoke(f"--batch task collect {PACKAGE_PATH}")
     debug(res0)
-    res0.show()
 
-    venv_path = res0.data["data"]["venv_path"]
-    state_id = res0.data["id"]
-    debug(venv_path)
-    debug(state_id)
+    state_id = res0.data[0]  # extract id from batch string
+    assert res0.data == "1 .fractal/fractal-tasks-mock0.0.1"
 
     time.sleep(0.5)
 
@@ -128,343 +122,6 @@ def test_repeated_task_collection(register_user, invoke, testdata_path):
         assert time.perf_counter() - starting_time < COLLECTION_TIMEOUT
 
     # Second collection
-    res0 = invoke(f"task collect {PACKAGE}")
-    res0.show()
+    res0 = invoke(f"task collect {PACKAGE_PATH}")
+    debug(res0.data)
     assert res0.data["data"]["info"] == "Already installed"
-
-
-def test_task_new(register_user, invoke, tmp_path):
-
-    # create a new task with just positional required args
-
-    schema_path = str(tmp_path / "schema.json")
-    schema = dict(key1=1, key2=[2, "3"])
-    with open(schema_path, "w") as f:
-        json.dump(schema, f)
-
-    res = invoke(
-        "task new _name _command _source --version _version "
-        f"--args-schema {schema_path} --args-schema-version 1.0.0"
-    )
-    res.show()
-    assert res.retcode == 0
-    assert res.data["name"] == "_name"
-    assert res.data["command"] == "_command"
-    assert res.data["source"] == f"{register_user['username']}:_source"
-    assert res.data["input_type"] == res.data["output_type"] == "Any"
-    assert res.data["version"] == "_version"
-    assert res.data["meta"] == {}
-    assert res.data["args_schema"] == schema
-    assert res.data["args_schema_version"] == "1.0.0"
-
-    assert "owner" in res.data.keys()
-    first_task_id = int(res.data["id"])
-
-    # create a new task with batch option
-    res = invoke("--batch task new _name2 _command2 _source2")
-    res.show()
-    assert res.retcode == 0
-    assert res.data == str(first_task_id + 1)
-
-    # create a new task with same source as before. Note that in check_response
-    # we have sys.exit(1) when status code is not the expecte one
-    with pytest.raises(SystemExit) as e:
-        invoke("task new _name2 _command2 _source")
-    assert e.value.code == 1
-
-    # create a new task passing not existing file
-    res = invoke("task new _name _command _source --meta-file ./foo.pdf")
-    debug(res.data)
-    assert res.retcode == 1
-
-
-def test_task_edit(
-    caplog,
-    register_user,
-    invoke,
-    invoke_as_superuser,
-    tmp_path,
-    testdata_path: Path,
-):
-    schema_path = str(tmp_path / "schema.json")
-    schema = dict(key1=1, key2=[2, "3"])
-    with open(schema_path, "w") as f:
-        json.dump(schema, f)
-
-    task = invoke(
-        "task new _name _command _source "
-        f"--args-schema {schema_path} --args-schema-version 1.0.0"
-    )
-    task.show()
-    assert task.retcode == 0
-    task_id = task.data["id"]
-    NEW_NAME = "1234"
-
-    # Test that regular user is not authorized
-    with pytest.raises(SystemExit):
-        res = invoke(f"task edit {task_id} --new-name {NEW_NAME}")
-
-    # Test successful edit of string attributes
-    res = invoke_as_superuser(
-        f"task edit --id {task_id} --new-name {NEW_NAME}"
-    )
-    assert res.data["name"] == NEW_NAME
-    assert res.retcode == 0
-
-    NEW_COMMAND = "run"
-    res = invoke_as_superuser(
-        f"task edit --id {task_id} --new-command {NEW_COMMAND}"
-    )
-    assert res.data["command"] == NEW_COMMAND
-    assert res.retcode == 0
-
-    # Test fail with no task_id nor task_name
-    with pytest.raises(SystemExit):
-        res = invoke_as_superuser("task edit")
-    # Test fail with both task_id and task_name
-    with pytest.raises(SystemExit):
-        res = invoke_as_superuser(
-            f"task edit --id {task_id} --name {task.data['name']}"
-        )
-    # Test fail with both task_id and task_version
-    with pytest.raises(SystemExit):
-        res = invoke_as_superuser(
-            f"task edit --id {task_id} --version 1.2.3.4.5.6"
-        )
-    assert caplog.records[-1].msg == (
-        "Too many arguments: cannot provide both `id` and `version`."
-    )
-
-    NEW_TYPE = "zip"
-    # Test regular updates (both by id and name)
-    res = invoke_as_superuser(
-        f"task edit --id {task_id} --new-input-type {NEW_TYPE}"
-    )
-    assert res.data["input_type"] == NEW_TYPE
-    assert res.retcode == 0
-    res = invoke_as_superuser(
-        f"task edit --name {NEW_NAME} --new-output-type {NEW_TYPE}"
-    )
-    assert res.data["output_type"] == NEW_TYPE
-    assert res.retcode == 0
-
-    NEW_VERSION = "3.14"
-    res = invoke_as_superuser(
-        f"task edit --id {task_id} --new-version {NEW_VERSION}"
-    )
-    assert res.data["version"] == NEW_VERSION
-    assert res.retcode == 0
-
-    # Test regular update by name, after deleting cache
-    cache_dir = Path(settings.FRACTAL_CACHE_PATH)
-    cache_file = cache_dir / TASKS_CACHE_FILENAME
-    cache_file.unlink(missing_ok=True)
-    NEW_TYPE = "something"
-    res = invoke_as_superuser(
-        f"task edit --name {NEW_NAME} --new-output-type {NEW_TYPE}"
-    )
-    assert res.data["output_type"] == NEW_TYPE
-    assert res.retcode == 0
-
-    # Test failed update by name, after deleting cache
-    cache_file.unlink(missing_ok=True)
-    NEW_TYPE = "something-here"
-    with pytest.raises(SystemExit):
-        res = invoke_as_superuser(
-            f"task edit --name INVALID_NAME --new-output-type {NEW_TYPE}"
-        )
-
-    # Test regular update by name, after creating an invalid cache
-    with cache_file.open("w") as f:
-        json.dump([], f)
-    NEW_TYPE = "something-else"
-    debug(f"task edit --name {NEW_NAME} --new-output-type {NEW_TYPE}")
-    res = invoke_as_superuser(
-        f"task edit --name {NEW_NAME} --new-output-type {NEW_TYPE}"
-    )
-    assert res.data["output_type"] == NEW_TYPE
-    assert res.retcode == 0
-
-    # Test `file not found` errors
-    NEW_FILE = "foo.json"
-    res = invoke_as_superuser(
-        f"task edit --id {task_id} --meta-file {NEW_FILE}"
-    )
-    debug(res.data)
-    assert res.retcode == 1
-
-    # Test successful edit of dictionary attributes
-    meta_file = str(testdata_path / "task_edit_json/meta.json")
-    res = invoke_as_superuser(
-        f"task edit --id {task_id} --meta-file {meta_file}"
-    )
-    debug(res)
-    debug(res.data)
-    assert res.retcode == 0
-
-    # Test succesful edit of args_schema (and args_schema_version)
-    new_schema_path = str(tmp_path / "new_schema.json")
-    new_schema = dict(key1=159, key3=None)
-    with open(new_schema_path, "w") as f:
-        json.dump(new_schema, f)
-    res = invoke_as_superuser(
-        f"task edit --id {task_id} --new-args-schema {new_schema_path} "
-        "--new-args-schema-version 1.2.3"
-    )
-    assert res.retcode == 0
-    assert res.data["args_schema"] == new_schema
-    assert res.data["args_schema_version"] == "1.2.3"
-
-
-def test_task_delete(
-    register_user,
-    user_factory,
-    invoke,
-    invoke_as_superuser,
-):
-    """
-    Test task delete
-    """
-    NAME = "_name"
-    VERSION = "1.0.0"
-    task = invoke(f"task new {NAME} _command _source --version {VERSION}")
-    task.show()
-    assert task.retcode == 0
-    task_id = task.data["id"]
-
-    # Test access control
-    with pytest.raises(SystemExit):
-        EMAIL = "someone@example.org"
-        PASSWORD = "123123"
-        user_factory(email=EMAIL, password=PASSWORD)
-        res = invoke(f"-u {EMAIL} -p {PASSWORD} task delete --id {task_id}")
-    # Test fail "id and version"
-    with pytest.raises(SystemExit):
-        invoke(f"task delete --id {task_id} --version {VERSION}")
-    # Test fail "name and wrong version"
-    with pytest.raises(SystemExit):
-        invoke(f"task delete --name {NAME} --version INVALID_VERSION")
-
-    # Test sucess
-    res = invoke("task list")
-    task_list = res.data
-    assert len(task_list) == 1
-    res = invoke(f"task delete --name {NAME} --version {VERSION}")
-    assert res.retcode == 0
-    res = invoke("task list")
-    task_list = res.data
-    assert len(task_list) == 0
-
-
-def test_task_list(register_user, invoke, testdata_path):
-    """
-    Install tasks from a package
-    Add a custom task with an owner
-    List tasks in the appropriate way
-    """
-
-    for version in ["0.1.0", "0.2.0"]:
-        # Task collection
-        PACKAGE = (
-            testdata_path / f"fractal_tasks_dummy-{version}-py3-none-any.whl"
-        )
-
-        res = invoke(f"--batch task collect {PACKAGE}")
-        assert res.retcode == 0
-
-        state_id = res.data.split(" ")[0]
-
-        # Wait until collection is complete
-        time.sleep(1)
-        starting_time = time.perf_counter()
-        while True:
-            res = invoke(f"task check-collection {state_id}")
-            assert res.retcode == 0
-            if res.data["data"]["status"] == "OK":
-                break
-            assert time.perf_counter() - starting_time < COLLECTION_TIMEOUT
-            time.sleep(1)
-
-    # Add custom task
-    custom_task_name = "custom_task_name"
-    custom_task_command = "custom_task_command"
-    custom_task_source = "custom_task_source"
-    custom_task_version = "9.9"
-    custom_task_meta = testdata_path / "task_edit_json/meta.json"
-    res = invoke(
-        f"task new {custom_task_name} {custom_task_command} "
-        f"{custom_task_source} --meta-file {custom_task_meta} "
-        f"--version {custom_task_version}"
-    )
-    debug(res)
-    assert res.retcode == 0
-
-    # List tasks
-    res = invoke("task list")
-    assert res.retcode == 0
-    task_list = res.data
-
-    # Remove some attributes, to de-clutter output
-    for task in task_list:
-        for key in [
-            "command",
-            "meta",
-            "input_type",
-            "output_type",
-        ]:
-            if key in task:
-                task.pop(key)
-    debug(task_list)
-
-    # Check that tasks are sorted as expected
-    assert task_list[0]["name"] == "dummy"
-    assert task_list[0]["version"] == "0.1.0"
-    assert task_list[0]["owner"] is None
-    assert task_list[1]["name"] == "dummy"
-    assert task_list[1]["version"] == "0.2.0"
-    assert task_list[1]["owner"] is None
-    assert task_list[2]["name"] == "dummy parallel"
-    assert task_list[2]["version"] == "0.1.0"
-    assert task_list[2]["owner"] is None
-    assert task_list[3]["name"] == "dummy parallel"
-    assert task_list[3]["version"] == "0.2.0"
-    assert task_list[3]["owner"] is None
-    assert task_list[4]["name"] == custom_task_name
-    assert task_list[4]["version"] == custom_task_version
-    assert task_list[4]["owner"] is not None
-
-
-def test_pin(register_user, invoke, testdata_path, caplog):
-
-    PACKAGE = "fractal_tasks_core_alpha-0.0.1a0-py3-none-any.whl"
-    PIN = "pydantic=1.10.3"
-
-    with pytest.raises(SystemExit):
-        invoke(
-            f"task collect {testdata_path / PACKAGE} "
-            f"--pinned-dependency {PIN.replace('=','~')}"
-        )
-    assert "Pins must be written as" in caplog.records[-1].msg
-
-    res = invoke(
-        f"task collect {testdata_path / PACKAGE} --pinned-dependency {PIN}"
-    )
-    assert res.retcode == 0
-    state_id = res.data["id"]
-    starting_time = time.perf_counter()
-    while True:
-        res = invoke(f"task check-collection {state_id}")
-        time.sleep(1)
-        if res.data["data"]["status"] == "OK":
-            break
-        assert time.perf_counter() - starting_time < COLLECTION_TIMEOUT
-
-    res = invoke(f"task check-collection {state_id}")
-    assert res.retcode == 0
-    assert (
-        f"Currently installed version of {PIN.split('=')[0]}"
-        in res.data["data"]["log"]
-    ) and (
-        f"differs from pinned version ({PIN.split('=')[1]})"
-        in res.data["data"]["log"]
-    )
diff --git a/tests/test_workflow.py b/tests/test_workflow.py
index c48a6e04e..966e3906c 100644
--- a/tests/test_workflow.py
+++ b/tests/test_workflow.py
@@ -1,6 +1,5 @@
 import json
 import logging
-import time
 from pathlib import Path
 
 import pytest  # noqa F401
@@ -158,17 +157,33 @@ def test_workflow_add_task(
     res = invoke("project new MyProject")
     project_id = res.data["id"]
     wf = workflow_factory(project_id=project_id)
-    t = task_factory()
+    t = task_factory(type="parallel")
 
-    # Add a WorkflowTask with --args-file and --meta-file arguments
-    ARGS = {"arg": "arg_value"}
-    META = {"executor": "some-executor"}
-    args_file = tmp_path / "args_file.json"
-    with args_file.open("w") as f:
-        json.dump(ARGS, f)
-    meta_file = tmp_path / "meta_file.json"
-    with meta_file.open("w") as f:
-        json.dump(META, f)
+    INPUT_FILTERS = {"attributes": {"a": 1}, "types": {"b": True}}
+    ARGS_PARALLEL = {"image_dir": "/asdasd"}
+    ARGS_NON_PARALLEL = {"image_dir": "/dsadsa"}
+    META_PARALLEL = {"a": "b"}
+    META_NON_PARALLEL = {"c": "d"}
+
+    input_filters_file = tmp_path / "input_filters.json"
+    with input_filters_file.open("w") as f:
+        json.dump(INPUT_FILTERS, f)
+
+    args_parallel_file = tmp_path / "args_parallel_file.json"
+    with args_parallel_file.open("w") as f:
+        json.dump(ARGS_PARALLEL, f)
+
+    args_non_parallel_file = tmp_path / "args_non_parallel_file.json"
+    with args_non_parallel_file.open("w") as f:
+        json.dump(ARGS_NON_PARALLEL, f)
+
+    meta_parallel_file = tmp_path / "meta_paral.json"
+    with meta_parallel_file.open("w") as f:
+        json.dump(META_PARALLEL, f)
+
+    meta_non_parallel_file = tmp_path / "meta_non_paral.json"
+    with meta_non_parallel_file.open("w") as f:
+        json.dump(META_NON_PARALLEL, f)
 
     cmd = f"workflow add-task {project_id} {wf.id}"
     # Test fail with no task_id nor task_name
@@ -176,40 +191,90 @@ def test_workflow_add_task(
         invoke(cmd)
     # Test fail with both task_id and task_name
     with pytest.raises(SystemExit):
-        invoke(f"{cmd} --task-id {t.id} --task-name {t.name}")
+        invoke(
+            (
+                f"{cmd} --task-id {t.id} --task-name {t.name} "
+                f"--args-parallel {args_parallel_file}"
+            )
+        )
     # Test fail with both task_id and version
     with pytest.raises(SystemExit):
-        invoke(f"{cmd} --task-id {t.id} --task-version 1.2.3.4.5.6")
+        invoke(
+            (
+                f"{cmd} --task-id {t.id} --task-version 1.2.3.4.5.6 "
+                f"--args-parallel {args_parallel_file}"
+            )
+        )
     assert caplog.records[-1].msg == (
         "Too many arguments: cannot provide both `task_id` and `task_version`."
     )
 
-    cmd = (
-        f"{cmd} --task-id {t.id} "
-        f"--args-file {args_file} --meta-file {meta_file}"
+    cmd_args = (
+        f"{cmd} --task-id {t.id} --input-filters {input_filters_file} "
+        f"--args-parallel {args_parallel_file} "
     )
-    debug(cmd)
+    debug(cmd_args)
     # Test success
-    res = invoke(cmd)
+    res = invoke(cmd_args)
+    debug(res.data)
     assert res.retcode == 0
 
     workflow_task = res.data
     workflow_task_id_1 = workflow_task["id"]
     debug(workflow_task)
-    assert workflow_task["args"] == ARGS
-    assert workflow_task["meta"] == META
+    assert workflow_task["input_filters"] == INPUT_FILTERS
+    assert workflow_task["args_parallel"] == ARGS_PARALLEL
 
     # Add a WorkflowTask by Task.name with the --batch option
-    cmd = (
+    cmd_batch = (
         f"--batch workflow add-task {project_id} {wf.id} "
-        f"--task-name {t.name} --order 1"
+        f"--task-name {t.name} --order 1 --args-parallel {args_parallel_file}"
     )
-    debug(cmd)
-    res = invoke(cmd)
+    debug(cmd_batch)
+    res = invoke(cmd_batch)
     assert res.retcode == 0
     debug(res.data)
     workflow_task_id_2 = int(res.data)
 
+    # Add a WorkflowTask with meta-parallel args
+    cmd_meta = (
+        f"{cmd} --task-id {t.id} --input-filters {input_filters_file} "
+        f"--args-parallel {args_parallel_file} "
+        f"--meta-parallel {meta_parallel_file} "
+    )
+    debug(cmd_meta)
+    # Test success
+    res = invoke(cmd_meta)
+    debug(res.data)
+    assert res.retcode == 0
+
+    workflow_task = res.data
+    workflow_task_id_3 = workflow_task["id"]
+    assert workflow_task["meta_parallel"] == META_PARALLEL
+    assert workflow_task["args_parallel"] == ARGS_PARALLEL
+
+    # Add a WorkflowTask with meta-non-parallel args
+    t_non_parallel = task_factory(
+        type="non_parallel", source="source non_parallel"
+    )
+
+    cmd_meta = (
+        f"{cmd} --task-id {t_non_parallel.id} "
+        f"--input-filters {input_filters_file} "
+        f"--args-non-parallel {args_non_parallel_file} "
+        f"--meta-non-parallel {meta_non_parallel_file}"
+    )
+    debug(cmd_meta)
+    # Test success
+    res = invoke(cmd_meta)
+    debug(res.data)
+    assert res.retcode == 0
+
+    workflow_task = res.data
+    workflow_task_id_4 = workflow_task["id"]
+    assert workflow_task["meta_non_parallel"] == META_NON_PARALLEL
+    assert workflow_task["args_non_parallel"] == ARGS_NON_PARALLEL
+
     # Check that the WorkflowTask's in Workflow.task_list have the correct IDs
     cmd = f"workflow show {project_id} {wf.id}"
     res = invoke(cmd)
@@ -217,7 +282,12 @@ def test_workflow_add_task(
     workflow = res.data
     debug(workflow)
     list_IDs = [wftask["id"] for wftask in workflow["task_list"]]
-    assert list_IDs == [workflow_task_id_1, workflow_task_id_2]
+    assert list_IDs == [
+        workflow_task_id_1,
+        workflow_task_id_2,
+        workflow_task_id_3,
+        workflow_task_id_4,
+    ]
 
 
 def test_workflow_add_task_by_name(
@@ -236,10 +306,19 @@ def test_workflow_add_task_by_name(
     res = invoke("project new MyProject")
     project_id = res.data["id"]
     wf = workflow_factory(project_id=project_id)
-    task = task_factory()
+    task = task_factory(type="parallel")
     debug(task)
 
-    cmd = f"workflow add-task {project_id} {wf.id} --task-name {task.name}"
+    ARGS = {"image_dir": "/asdasd"}
+
+    args_file = tmp_path / "args_file.json"
+    with args_file.open("w") as f:
+        json.dump(ARGS, f)
+
+    cmd = (
+        f"workflow add-task {project_id} {wf.id} --task-name {task.name} "
+        f"--args-parallel {args_file}"
+    )
     debug(cmd)
     res = invoke(cmd)
     assert res.retcode == 0
@@ -249,7 +328,8 @@ def test_workflow_add_task_by_name(
     # Fail when adding task via a wrong name
     with pytest.raises(SystemExit):
         cmd = (
-            f"workflow add-task {project_id} {wf.id} --task-name INVALID_NAME"
+            f"workflow add-task {project_id} {wf.id} --task-name INVALID_NAME "
+            f"--args-parallel {args_file}"
         )
         debug(cmd)
         res = invoke(cmd)
@@ -274,10 +354,14 @@ def test_task_cache_with_non_unique_names(
 
     res = invoke("project new MyProject")
     project_id = res.data["id"]
+    ARGS = {"image_dir": "/asdasd"}
 
+    args_file = tmp_path / "args_file.json"
+    with args_file.open("w") as f:
+        json.dump(ARGS, f)
     # Create two tasks with the same name
-    task1 = task_factory()
-    task2 = task_factory()
+    task1 = task_factory(type="parallel")
+    task2 = task_factory(type="parallel")
     assert task1.name == task2.name
 
     # Verify that a warning is raised upon creating the cache file
@@ -290,7 +374,10 @@ def test_task_cache_with_non_unique_names(
     # Verify that adding tasks to a worfklow by name (as opposed to "by id")
     # fails because of missing cache file
     wf = workflow_factory(project_id=project_id)
-    cmd = f"workflow add-task {project_id} {wf.id} --task-name {task1.name}"
+    cmd = (
+        f"workflow add-task {project_id} {wf.id} --task-name {task1.name} "
+        f"--args-parallel {args_file}"
+    )
     debug(cmd)
     with pytest.raises(FileNotFoundError):
         res = invoke(cmd)
@@ -307,10 +394,19 @@ def test_workflow_rm_task(
     res = invoke("project new MyProject")
     project_id = res.data["id"]
     wf = workflow_factory(project_id=project_id)
-    t = task_factory()
+    t = task_factory(type="parallel")
+
+    ARGS = {"image_dir": "/asdasd"}
+
+    args_file = tmp_path / "args_file.json"
+    with args_file.open("w") as f:
+        json.dump(ARGS, f)
 
     # Add task to workflow, twice
-    cmd = f"workflow add-task {project_id} {wf.id} --task-id {t.id}"
+    cmd = (
+        f"workflow add-task {project_id} {wf.id} --task-id {t.id} "
+        f"--args-parallel {args_file}"
+    )
     res = invoke(cmd)
     assert res.retcode == 0
     res = invoke(cmd)
@@ -343,327 +439,94 @@ def test_workflow_edit_task(
     res = invoke("project new MyProject")
     project_id = res.data["id"]
     wf = workflow_factory(project_id=project_id)
-    t = task_factory()
+    t = task_factory(type="parallel")
+
+    INPUT_FILTERS = {"attributes": {"a": 1}, "types": {"b": True}}
+    ARGS_PARALLEL = {"image_dir": "/asdasd"}
+    ARGS_NON_PARALLEL = {"image_dir": "/dsadsa"}
+    META_PARALLEL = {"a": "b"}
+    META_NON_PARALLEL = {"c": "d"}
+
+    input_filters_file = tmp_path / "input_filters.json"
+    with input_filters_file.open("w") as f:
+        json.dump(INPUT_FILTERS, f)
+
+    args_parallel_file = tmp_path / "args_parallel_file.json"
+    with args_parallel_file.open("w") as f:
+        json.dump(ARGS_PARALLEL, f)
+
+    args_non_parallel_file = tmp_path / "args_non_parallel_file.json"
+    with args_non_parallel_file.open("w") as f:
+        json.dump(ARGS_NON_PARALLEL, f)
+
+    meta_parallel_file = tmp_path / "meta_paral.json"
+    with meta_parallel_file.open("w") as f:
+        json.dump(META_PARALLEL, f)
+
+    meta_non_parallel_file = tmp_path / "meta_non_paral.json"
+    with meta_non_parallel_file.open("w") as f:
+        json.dump(META_NON_PARALLEL, f)
 
     # Create task, without overriding arguments
-    cmd = f"workflow add-task {project_id} {wf.id} --task-id {t.id}"
+    cmd = (
+        f"workflow add-task {project_id} {wf.id} --task-id {t.id} "
+        f"--args-parallel {args_parallel_file}"
+    )
     res = invoke(cmd)
     assert res.retcode == 0
 
-    # New arguments to be used
-    ARGS = {"some_arg": "some_value"}
-    META = {"executor": "cpu-low"}
-
-    args_file = tmp_path / "args_file.json"
-    with args_file.open("w") as f:
-        json.dump(ARGS, f)
-    meta_file = tmp_path / "meta_file.json"
-    with meta_file.open("w") as f:
-        json.dump(META, f)
-
     # Edit workflow task
     debug(res.data)
     workflow_task_id = res.data["id"]
     cmd = (
         f"workflow edit-task {project_id} {wf.id} {workflow_task_id} "
-        f"--args-file {args_file} --meta-file {meta_file}"
+        f"--input-filters {input_filters_file}"
     )
     debug(cmd)
     res = invoke(cmd)
     assert res.retcode == 0
-    assert res.data["args"] == ARGS
-    assert res.data["meta"] == META
+    assert res.data["input_filters"] == INPUT_FILTERS
 
-    # Check that also the workflow in the db was correctly updated
-    res = invoke(f"workflow show {project_id} {wf.id}")
-    assert res.retcode == 0
+    # Edit workflow task
     debug(res.data)
-    assert res.data["task_list"][0]["args"] == ARGS
-    assert res.data["task_list"][0]["meta"] == META
-
-    # Check if the correct error is raised where parallelization_level
-    # is set
-    META_err = {"parallelization_level": "XXX"}
-    meta_file = tmp_path / "meta_file.json"
-    with meta_file.open("w") as f:
-        json.dump(META_err, f)
-
-    workflow_task_id = res.data["task_list"][0]["id"]
+    workflow_task_id = res.data["id"]
     cmd = (
         f"workflow edit-task {project_id} {wf.id} {workflow_task_id} "
-        f"--meta-file {meta_file}"
-    )
-    debug(cmd)
-    with pytest.raises(SystemExit):
-        res = invoke(cmd)
-
-
-def test_workflow_apply(
-    register_user, invoke, testdata_path: Path, tmp_path: Path
-):
-    """
-    GIVEN a project and a nontrivial workflow
-    WHEN the client requests to apply the workflow to the project
-    THEN the workflow is scheduled and executed, and the artifacts created
-    """
-
-    # Collect tasks
-    PACKAGE_NAME = (
-        testdata_path / "fractal_tasks_core_alpha-0.0.1a0-py3-none-any.whl"
-    )
-    WORKFLOW_NAME = "mywf"
-    DATASET_NAME = "myds"
-    res0 = invoke(f"task collect {PACKAGE_NAME}")
-    debug(res0)
-    venv_path = res0.data["data"]["venv_path"]
-    debug(venv_path)
-    state_id = res0.data["id"]
-
-    # Wait for task collection to end
-    starting_time = time.perf_counter()
-    while True:
-        res1 = invoke(f"task check-collection {state_id}")
-        if res1.data["data"]["status"] == "OK":
-            debug(res1.data)
-            break
-        time.sleep(1)
-        assert time.perf_counter() - starting_time < TIMEOUT
-
-    # Create a project
-    res = invoke("project new testproject")
-    assert res.retcode == 0
-    prj = res.data
-    prj_id = prj["id"]
-    res = invoke(f"project add-dataset {prj_id} test_name")
-    input_dataset_id = res.data["id"]
-
-    # Create output dataset
-    res = invoke(f"project add-dataset {prj_id} {DATASET_NAME}")
-    assert res.retcode == 0
-    dataset = res.data
-    output_dataset_id = dataset["id"]
-
-    # Add resources to datasets
-    for dataset_id in (input_dataset_id, output_dataset_id):
-        res = invoke(
-            f"dataset add-resource {prj_id} {dataset_id} {str(tmp_path)}"
-        )
-        assert res.retcode == 0
-
-    # Create workflow and add task twice
-    res = invoke(f"workflow new {WORKFLOW_NAME} {prj_id}")
-    workflow = res.data
-    workflow_id = workflow["id"]
-    debug(workflow)
-    assert res.retcode == 0
-    for _ in [0, 1]:
-        TASK_ID = 1
-        res = 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"--start {FIRST_TASK_INDEX} --end {LAST_TASK_INDEX} "
-        f'--worker-init "export SOMEVARIABLE=1"'
+        f"--args-parallel {args_parallel_file} "
+        f"--meta-parallel {meta_parallel_file}"
     )
     debug(cmd)
     res = invoke(cmd)
-    job = res.data
-    debug(job)
     assert res.retcode == 0
-    job_id = job["id"]
-    assert job["status"] == "submitted"
-
-    # Avoid immediately calling `job show` right after `workflow apply`
-    time.sleep(1)
+    assert res.data["meta_parallel"] == META_PARALLEL
+    assert res.data["args_parallel"] == ARGS_PARALLEL
 
-    # Check that job completed successfully
-    cmd = f"job show {prj_id} {job_id} --do-not-separate-logs"
-    starting_time = time.perf_counter()
-    debug(cmd)
-    while True:
-        res = invoke(cmd)
-        job = res.data
-        debug(job)
-        assert res.retcode == 0
-        if job["status"] == "done":
-            break
-        elif job["status"] == "failed":
-            raise RuntimeError(job)
-        time.sleep(1)
-        assert time.perf_counter() - starting_time < TIMEOUT
-
-    # 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:
-        json.dump({"raise_error": True}, f)
-    res = invoke(
-        f"workflow add-task {prj_id} {workflow_id} --task-id {TASK_ID}"
-        f" --args-file {args_file}"
+    # Add a WorkflowTask with meta-non-parallel args
+    t_non_parallel = task_factory(
+        type="non_parallel", source="source non_parallel"
     )
-    assert res.retcode == 0
+
     cmd = (
-        f"workflow apply "
-        f"{prj_id} {workflow_id} {input_dataset_id} {output_dataset_id}"
+        f"workflow add-task {project_id} {wf.id} "
+        f"--task-id {t_non_parallel.id} "
+        f"--args-non-parallel {args_non_parallel_file}"
     )
-    debug(cmd)
     res = invoke(cmd)
     assert res.retcode == 0
-    job_id = res.data["id"]
-
-    # Avoid immediately calling `job show` right after `workflow apply`
-    time.sleep(1)
+    workflow_task_id = res.data["id"]
 
-    # Verify that status is failed, and that there is a log
-    cmd = f"job show {prj_id} {job_id} --do-not-separate-logs"
-    starting_time = time.perf_counter()
-    while True:
-        res = invoke(cmd)
-        job = res.data
-        debug(job)
-        assert res.retcode == 0
-        if job["status"] == "failed":
-            break
-        time.sleep(1)
-        assert time.perf_counter() - starting_time < TIMEOUT
-    assert job["log"] is not None
-
-    # Prepare and submit a workflow with --batch
-    res = invoke(f"workflow new OneMoreWorkflow {prj_id}")
-    workflow_id = res.data["id"]
-    res = invoke(
-        f"workflow add-task {prj_id} {workflow_id} --task-id {TASK_ID}"
-    )
-    assert res.retcode == 0
     cmd = (
-        f"--batch workflow apply "
-        f"{prj_id} {workflow_id} {input_dataset_id} {output_dataset_id}"
+        f"workflow edit-task {project_id} {wf.id} {workflow_task_id} "
+        f"--input-filters {input_filters_file} "
+        f"--args-non-parallel {args_non_parallel_file} "
+        f"--meta-non-parallel {meta_non_parallel_file}"
     )
     debug(cmd)
+    # Test success
     res = invoke(cmd)
-    assert res.retcode == 0
-    debug(res.data)
-    assert isinstance(res.data, int)
-
-
-def test_workflow_export(
-    register_user,
-    invoke,
-    workflow_factory,
-    tmp_path: Path,
-    task_factory,
-    caplog,
-):
-
-    res = invoke("project new testproject")
-    assert res.retcode == 0
-    project_id = res.data["id"]
-
-    NAME = "WorkFlow"
-    wf = workflow_factory(project_id=project_id, name=NAME)
-    prj_id = wf.project_id
-    wf_id = wf.id
-    filename = str(tmp_path / "exported_wf.json")
-
-    task = task_factory(owner="exact-lab")
-    res = invoke(f"workflow add-task {prj_id} {wf_id} --task-id {task.id}")
-    assert res.retcode == 0
-
-    res = invoke(f"workflow export {prj_id} {wf_id} --json-file {filename}")
-    assert res.retcode == 0
-    assert caplog.records[-1].msg == (
-        "This workflow includes custom tasks (the ones with sources: "
-        f"'{task.source}'), which are not meant to be portable; "
-        "re-importing this workflow may not work as expected."
-    )
-    debug(res.data)
-    with open(filename, "r") as f:
-        exported_wf = json.load(f)
-        assert exported_wf["name"] == NAME
-        assert "id" not in exported_wf
-        assert "project_id" not in exported_wf
-        for wftask in exported_wf["task_list"]:
-            assert "id" not in wftask
-            assert "task_id" not in wftask
-            assert "workflow_id" not in wftask
-
-
-def test_workflow_import(
-    register_user,
-    invoke,
-    testdata_path: Path,
-    task_factory,
-    caplog,
-):
-    # collect tasks
-    PACKAGE_NAME = testdata_path / "fractal_tasks_dummy-0.1.0-py3-none-any.whl"
-    res0 = invoke(f"task collect {PACKAGE_NAME}")
-    assert res0.retcode == 0
-    state_id = res0.data["id"]
-    starting_time = time.perf_counter()
-    while True:
-        res1 = invoke(f"task check-collection {state_id}")
-        if res1.data["data"]["status"] == "OK":
-            break
-        time.sleep(1)
-        assert time.perf_counter() - starting_time < TIMEOUT
-
-    # create project
-    PROJECT_NAME = "project_name"
-    res_pj = invoke(f"project new {PROJECT_NAME}")
-    assert res_pj.retcode == 0
-    project_id = res_pj.data["id"]
-
-    task_factory(source="custom_source", owner="exact-lab")
-
-    # import workflow into project
-    filename = str(testdata_path / "import-export/workflow.json")
-    res = invoke(
-        f"workflow import --project-id {project_id} --json-file {filename}"
-    )
     debug(res.data)
     assert res.retcode == 0
-    assert caplog.records[-1].msg == (
-        "This workflow includes custom tasks (the ones with sources: "
-        "'custom_source'), which are not meant to be portable; "
-        "importing this workflow may not work as expected."
-    )
-
-    imported_workflow = res.data
 
-    # get the workflow from the server, and check that it is the same
-    workflow_id = res.data["id"]
-    res = invoke(f"workflow show {project_id} {workflow_id}")
-    assert res.retcode == 0
-    assert res.data == imported_workflow
-
-    # import workflow into project, with --batch
-    filename = str(testdata_path / "import-export/workflow_2.json")
-    res = invoke(
-        f"--batch workflow import --project-id {project_id} "
-        f"--json-file {filename}"
-    )
-    debug(res.data)
-    for _id in res.data.split():
-        assert _id.isdigit()
-    assert res.retcode == 0
-    assert caplog.records[-1].msg == (
-        "This workflow includes custom tasks (the ones with sources: "
-        "'custom_source'), which are not meant to be portable; "
-        "importing this workflow may not work as expected."
-    )
+    workflow_task = res.data
+    assert workflow_task["meta_non_parallel"] == META_NON_PARALLEL
+    assert workflow_task["args_non_parallel"] == ARGS_NON_PARALLEL