Deprecate use of v1 tasks within v2 workflows

CHANGELOG.md

@@ -1,5 +1,21 @@
 **Note**: Numbers like (\#1234) point to closed Pull Requests on the fractal-server repository.
 
+# 2.5.0
+
+This release removes support for including V1 tasks in V2 workflows. This comes
+with changes to the database (data and metadata), to the API, and to the V2
+runner.
+
+* Runner:
+    * Deprecate running v1 tasks within v2 workflows (\#1721).
+* Database:
+    * Remove `Task.is_v2_compatible` column (\#1721).
+    * For table `WorkflowTaskV2`, drop `is_legacy_task` and `task_legacy_id` columns, remove `task_legacy` ORM attribute, make `task_id` required, make `task` required (\#1721).
+* API:
+    * Drop v1-v2-task-compatibility admin endpoint (\#1721).
+    * Drop `/task-legacy/` endpoint (\#1721).
+    * Remove legacy task code branches from `WorkflowTaskV2` CRUD endpoints (\#1721).
+
 # 2.4.2
 
 * Runner:
@@ -18,6 +34,11 @@ This is mainly a bugfix release, re-implementing a check that was removed in 2.4
 
 # 2.4.0
 
+This release introduces support for user groups, but without linking it to any
+access-control rules (which will be introduced later).
+
+> NOTE: This release requires running the `fractalctl update-db-data` script.
+
 * App:
     * Move creation of first user from application startup into `fractalctl set-db` command (\#1738, \#1748).
     * Add creation of default user group into `fractalctl set-db` command (\#1738).

benchmarks/runner/mocks.py

@@ -69,59 +69,19 @@ def _set_type(cls, value, values):
                 return "compound"
 
 
-class TaskV1Mock(BaseModel):
-    id: int
-    name: str
-    command: str  # str
-    source: str = Field(unique=True)
-    input_type: str
-    output_type: str
-    meta: Optional[dict[str, Any]] = Field(default_factory=dict)
-
-    @property
-    def parallelization_level(self) -> Optional[str]:
-        try:
-            return self.meta["parallelization_level"]
-        except KeyError:
-            return None
-
-    @property
-    def is_parallel(self) -> bool:
-        return bool(self.parallelization_level)
-
-
 class WorkflowTaskV2Mock(BaseModel):
     args_non_parallel: dict[str, Any] = Field(default_factory=dict)
     args_parallel: dict[str, Any] = Field(default_factory=dict)
     meta_non_parallel: dict[str, Any] = Field(default_factory=dict)
     meta_parallel: dict[str, Any] = Field(default_factory=dict)
-    is_legacy_task: Optional[bool]
     meta_parallel: Optional[dict[str, Any]] = Field()
     meta_non_parallel: Optional[dict[str, Any]] = Field()
-    task: Optional[TaskV2Mock] = None
-    task_legacy: Optional[TaskV1Mock] = None
-    is_legacy_task: bool = False
+    task: TaskV2Mock = None
     input_filters: dict[str, Any] = Field(default_factory=dict)
     order: int
     id: int
     workflow_id: int = 0
-    task_legacy_id: Optional[int]
-    task_id: Optional[int]
-
-    @root_validator(pre=False)
-    def _legacy_or_not(cls, values):
-        is_legacy_task = values["is_legacy_task"]
-        task = values.get("task")
-        task_legacy = values.get("task_legacy")
-        if is_legacy_task:
-            if task_legacy is None or task is not None:
-                raise ValueError(f"Invalid WorkflowTaskV2Mock with {values=}")
-            values["task_legacy_id"] = task_legacy.id
-        else:
-            if task is None or task_legacy is not None:
-                raise ValueError(f"Invalid WorkflowTaskV2Mock with {values=}")
-            values["task_id"] = task.id
-        return values
+    task_id: int
 
     @validator("input_filters", always=True)
     def _default_filters(cls, value):

fractal_server/app/db/__init__.py

@@ -2,6 +2,7 @@
 `db` module, loosely adapted from
 https://testdriven.io/blog/fastapi-sqlmodel/#async-sqlmodel
 """
+import sqlite3
 from typing import AsyncGenerator
 from typing import Generator
 
@@ -21,7 +22,9 @@
 logger = set_logger(__name__)
 
 SQLITE_WARNING_MESSAGE = (
-    "SQLite is supported (for version >=3.37) but discouraged in production. "
+    "SQLite is supported (supported version >=3.37, "
+    f"current {sqlite3.sqlite_version=}) "
+    "but discouraged in production. "
     "Given its partial support for ForeignKey constraints, "
     "database consistency cannot be guaranteed."
 )

fractal_server/app/models/v1/task.py

@@ -5,7 +5,6 @@
 
 from pydantic import HttpUrl
 from sqlalchemy import Column
-from sqlalchemy import sql
 from sqlalchemy.types import JSON
 from sqlmodel import Field
 from sqlmodel import SQLModel
@@ -49,10 +48,6 @@ class Task(_TaskBaseV1, SQLModel, table=True):
     docs_info: Optional[str] = None
     docs_link: Optional[HttpUrl] = None
 
-    is_v2_compatible: bool = Field(
-        default=False, sa_column_kwargs={"server_default": sql.false()}
-    )
-
     @property
     def parallelization_level(self) -> Optional[str]:
         try:

fractal_server/app/models/v2/workflowtask.py

@@ -8,7 +8,6 @@
 from sqlmodel import Relationship
 from sqlmodel import SQLModel
 
-from ..v1.task import Task
 from .task import TaskV2
 
 
@@ -37,13 +36,6 @@ class Config:
     )
 
     # Task
-    is_legacy_task: bool
     task_type: str
-    task_id: Optional[int] = Field(foreign_key="taskv2.id")
-    task: Optional[TaskV2] = Relationship(
-        sa_relationship_kwargs=dict(lazy="selectin")
-    )
-    task_legacy_id: Optional[int] = Field(foreign_key="task.id")
-    task_legacy: Optional[Task] = Relationship(
-        sa_relationship_kwargs=dict(lazy="selectin")
-    )
+    task_id: int = Field(foreign_key="taskv2.id")
+    task: TaskV2 = Relationship(sa_relationship_kwargs=dict(lazy="selectin"))

fractal_server/app/routes/admin/v2.py

@@ -24,7 +24,6 @@
 from ....zip_tools import _zip_folder_to_byte_stream_iterator
 from ...db import AsyncSession
 from ...db import get_async_db
-from ...models.v1 import Task
 from ...models.v2 import JobV2
 from ...models.v2 import ProjectV2
 from ...models.v2 import TaskV2
@@ -281,35 +280,6 @@ async def download_job_logs(
     )
 
 
-class TaskCompatibility(BaseModel):
-    is_v2_compatible: bool
-
-
-@router_admin_v2.patch(
-    "/task-v1/{task_id}/",
-    status_code=status.HTTP_200_OK,
-)
-async def flag_task_v1_as_v2_compatible(
-    task_id: int,
-    compatibility: TaskCompatibility,
-    user: UserOAuth = Depends(current_active_superuser),
-    db: AsyncSession = Depends(get_async_db),
-) -> Response:
-
-    task = await db.get(Task, task_id)
-    if task is None:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND,
-            detail=f"Task {task_id} not found",
-        )
-
-    task.is_v2_compatible = compatibility.is_v2_compatible
-    await db.commit()
-    await db.close()
-
-    return Response(status_code=status.HTTP_200_OK)
-
-
 class TaskV2Minimal(BaseModel):
 
     id: int

fractal_server/app/routes/api/v2/__init__.py

@@ -12,7 +12,6 @@
 from .task import router as task_router_v2
 from .task_collection import router as task_collection_router_v2
 from .task_collection_custom import router as task_collection_router_v2_custom
-from .task_legacy import router as task_legacy_router_v2
 from .workflow import router as workflow_router_v2
 from .workflowtask import router as workflowtask_router_v2
 from fractal_server.config import get_settings
@@ -38,9 +37,6 @@
     tags=["V2 Task Collection"],
 )
 router_api_v2.include_router(task_router_v2, prefix="/task", tags=["V2 Task"])
-router_api_v2.include_router(
-    task_legacy_router_v2, prefix="/task-legacy", tags=["V2 Task Legacy"]
-)
 router_api_v2.include_router(workflow_router_v2, tags=["V2 Workflow"])
 router_api_v2.include_router(workflowtask_router_v2, tags=["V2 WorkflowTask"])
 router_api_v2.include_router(status_router_v2, tags=["V2 Status"])

fractal_server/app/routes/api/v2/_aux_functions.py

@@ -13,7 +13,6 @@
 from sqlmodel.sql.expression import SelectOfScalar
 
 from ....db import AsyncSession
-from ....models.v1 import Task
 from ....models.v2 import DatasetV2
 from ....models.v2 import JobV2
 from ....models.v2 import LinkUserProjectV2
@@ -389,7 +388,6 @@ async def _workflow_insert_task(
     *,
     workflow_id: int,
     task_id: int,
-    is_legacy_task: bool = False,
     order: Optional[int] = None,
     meta_parallel: Optional[dict[str, Any]] = None,
     meta_non_parallel: Optional[dict[str, Any]] = None,
@@ -404,7 +402,7 @@ async def _workflow_insert_task(
     Args:
         workflow_id:
         task_id:
-        is_legacy_task:
+
         order:
         meta_parallel:
         meta_non_parallel:
@@ -420,52 +418,21 @@ async def _workflow_insert_task(
     if order is None:
         order = len(db_workflow.task_list)
 
-    # Get task from db, and extract default arguments via a Task property
-    # method
-    # NOTE: this logic remains there for V1 tasks only. When we deprecate V1
-    # tasks, we can simplify this block
-    if is_legacy_task is True:
-        db_task = await db.get(Task, task_id)
-        if db_task is None:
-            raise ValueError(f"Task {task_id} not found.")
-        task_type = "parallel"
-
-        final_args_parallel = db_task.default_args_from_args_schema.copy()
-        final_args_non_parallel = {}
-        final_meta_parallel = (db_task.meta or {}).copy()
-        final_meta_non_parallel = {}
-
-    else:
-        db_task = await db.get(TaskV2, task_id)
-        if db_task is None:
-            raise ValueError(f"TaskV2 {task_id} not found.")
-        task_type = db_task.type
-
-        final_args_non_parallel = {}
-        final_args_parallel = {}
-        final_meta_parallel = (db_task.meta_parallel or {}).copy()
-        final_meta_non_parallel = (db_task.meta_non_parallel or {}).copy()
-
-    # Combine arg_parallel
-    if args_parallel is not None:
-        for k, v in args_parallel.items():
-            final_args_parallel[k] = v
-    if final_args_parallel == {}:
-        final_args_parallel = None
-    # Combine arg_non_parallel
-    if args_non_parallel is not None:
-        for k, v in args_non_parallel.items():
-            final_args_non_parallel[k] = v
-    if final_args_non_parallel == {}:
-        final_args_non_parallel = None
+    # Get task from db
+    db_task = await db.get(TaskV2, task_id)
+    if db_task is None:
+        raise ValueError(f"TaskV2 {task_id} not found.")
+    task_type = db_task.type
 
     # Combine meta_parallel (higher priority)
     # and db_task.meta_parallel (lower priority)
+    final_meta_parallel = (db_task.meta_parallel or {}).copy()
     final_meta_parallel.update(meta_parallel or {})
     if final_meta_parallel == {}:
         final_meta_parallel = None
     # Combine meta_non_parallel (higher priority)
     # and db_task.meta_non_parallel (lower priority)
+    final_meta_non_parallel = (db_task.meta_non_parallel or {}).copy()
     final_meta_non_parallel.update(meta_non_parallel or {})
     if final_meta_non_parallel == {}:
         final_meta_non_parallel = None
@@ -479,11 +446,9 @@ async def _workflow_insert_task(
     # Create DB entry
     wf_task = WorkflowTaskV2(
         task_type=task_type,
-        is_legacy_task=is_legacy_task,
-        task_id=(task_id if not is_legacy_task else None),
-        task_legacy_id=(task_id if is_legacy_task else None),
-        args_non_parallel=final_args_non_parallel,
-        args_parallel=final_args_parallel,
+        task_id=task_id,
+        args_non_parallel=args_non_parallel,
+        args_parallel=args_parallel,
         meta_parallel=final_meta_parallel,
         meta_non_parallel=final_meta_non_parallel,
         **input_filters_kwarg,