workflow_import

AvailableTask

Bases: BaseModel

Represents an alternative for a workflow task that one has attempted to import but which is not available.

Source code in fractal_server/app/routes/api/v2/workflow_import.py

class AvailableTask(BaseModel):
    """
    Represents an alternative for a workflow task that one has attempted to
    import but which is not available.
    """

    version: str
    older_than_target: bool
    active: bool

_get_task_id_or_available_tasks(*, task_import, task_groups_list, user_id, default_group_id, db) async

Find a task id based on task_import. If the task is not found, return the list of available versions.

PARAMETER	DESCRIPTION
task_import	Info on task to be imported. TYPE: TaskImport
task_groups_list	Current list of valid task groups with not-null version. TYPE: list[TaskGroupV2]
user_id	ID of current user. TYPE: int
default_group_id	ID of default user group. TYPE: int | None
db	Asynchronous database session. TYPE: AsyncSession

Return

A tuple (success, result) where: - success is True if a matching task was found, False otherwise. - result is: - the id of the matching task when success is True, - a list of AvailableTask instances when success is False.

Source code in fractal_server/app/routes/api/v2/workflow_import.py

async def _get_task_id_or_available_tasks(
    *,
    task_import: TaskImport,
    task_groups_list: list[TaskGroupV2],
    user_id: int,
    default_group_id: int | None,
    db: AsyncSession,
) -> tuple[bool, int | list[AvailableTask]]:
    """
    Find a task id based on `task_import`.
    If the task is not found, return the list of available versions.

    Args:
        task_import: Info on task to be imported.
        task_groups_list: Current list of valid task groups with not-null
            version.
        user_id: ID of current user.
        default_group_id: ID of default user group.
        db: Asynchronous database session.

    Return:
        A tuple `(success, result)` where:
        - `success` is `True` if a matching task was found, `False` otherwise.
        - `result` is:
            - the `id` of the matching task when `success` is `True`,
            - a list of `AvailableTask` instances when `success` is `False`.
    """

    logger.debug(f"[_get_task_id_or_available_tasks] START, {task_import=}")

    # Filter by `pkg_name` and by presence of a task with given `name`.
    matching_task_groups = [
        task_group
        for task_group in task_groups_list
        if (
            task_group.pkg_name == task_import.pkg_name
            and task_import.name in [task.name for task in task_group.task_list]
        )
    ]
    if len(matching_task_groups) < 1:
        logger.debug(
            "[_get_task_id_or_available_tasks] "
            f"No task group with {task_import.pkg_name=} "
            f"and a task with {task_import.name=}."
        )
        return (False, [])

    if task_import.version is not None:
        final_matching_task_groups = list(
            filter(
                lambda tg: tg.version == task_import.version,
                matching_task_groups,
            )
        )
    else:
        final_matching_task_groups = matching_task_groups

    if task_import.version is None or len(final_matching_task_groups) < 1:
        logger.debug(
            "[_get_task_id_or_available_tasks] "
            "No task group left after filtering by version."
        )
        return (
            False,
            [
                AvailableTask(
                    version=tg.version,
                    older_than_target=(
                        _version_sort_key(tg.version)
                        < _version_sort_key(task_import.version)
                    ),
                    active=tg.active,
                )
                for tg in matching_task_groups
            ],
        )
    elif len(final_matching_task_groups) == 1:
        final_task_group = final_matching_task_groups[0]
        logger.debug(
            "[_get_task_id_or_available_tasks] "
            "Found a single task group, after filtering by version."
        )
    else:
        logger.debug(
            "[_get_task_id_or_available_tasks] "
            f"Found {len(final_matching_task_groups)} task groups, "
            "after filtering by version."
        )
        final_task_group = await _disambiguate_task_groups(
            matching_task_groups=final_matching_task_groups,
            user_id=user_id,
            db=db,
            default_group_id=default_group_id,
        )
        if final_task_group is None:
            logger.error(
                "[_get_task_id_or_available_tasks] UnreachableBranchError: "
                "disambiguation returned None, likely be due to a race "
                "condition on TaskGroups."
            )
            return (False, [])

    # Find task with given name
    task_id = next(
        iter(
            task.id
            for task in final_task_group.task_list
            if task.name == task_import.name
        ),
        None,
    )
    if task_id is None:
        logger.error(
            "[_get_task_id_or_available_tasks] UnreachableBranchError: "
            "likely be due to a race condition on TaskGroups."
        )
        return (False, [])

    logger.debug(
        f"[_get_task_id_or_available_tasks] END, {task_import=}, {task_id=}."
    )

    return (True, task_id)

_get_user_accessible_taskgroups_with_version(*, user_id, user_resource_id, db) async

Retrieve list of task groups with non-null version that the user has access to.

Source code in fractal_server/app/routes/api/v2/workflow_import.py

async def _get_user_accessible_taskgroups_with_version(
    *,
    user_id: int,
    user_resource_id: int,
    db: AsyncSession,
) -> list[TaskGroupV2]:
    """
    Retrieve list of task groups with non-null version that the user has access
    to.
    """

    stm = (
        select(TaskGroupV2)
        .where(
            or_(
                TaskGroupV2.user_id == user_id,
                TaskGroupV2.user_group_id.in_(
                    select(LinkUserGroup.group_id).where(
                        LinkUserGroup.user_id == user_id
                    )
                ),
            )
        )
        .where(TaskGroupV2.resource_id == user_resource_id)
        .where(is_not(TaskGroupV2.version, None))
    )
    res = await db.execute(stm)
    accessible_task_groups = res.scalars().all()
    logger.debug(
        f"Found {len(accessible_task_groups)} accessible "
        f"task groups for {user_id=}."
    )
    return accessible_task_groups

_import_workflow(*, project_id, workflow_import, user, db, template_id=None) async

Import a workflow into a project and create required objects.

Source code in fractal_server/app/routes/api/v2/workflow_import.py

async def _import_workflow(
    *,
    project_id: int,
    workflow_import: WorkflowImport,
    user: UserOAuth,
    db: AsyncSession,
    template_id: int | None = None,
) -> dict[str, Any]:
    """
    Import a workflow into a project and create required objects.
    """
    user_resource_id = await _get_user_resource_id(user_id=user.id, db=db)

    # Preliminary checks
    await _get_project_check_access(
        project_id=project_id,
        user_id=user.id,
        required_permissions=ProjectPermissions.WRITE,
        db=db,
    )
    await _check_workflow_exists(
        name=workflow_import.name,
        project_id=project_id,
        db=db,
    )

    task_group_list = await _get_user_accessible_taskgroups_with_version(
        user_id=user.id,
        db=db,
        user_resource_id=user_resource_id,
    )
    default_group_id = await _get_default_usergroup_id_or_none(db)

    list_wf_tasks = []
    list_results = [
        await _get_task_id_or_available_tasks(
            task_import=wf_task.task,
            user_id=user.id,
            default_group_id=default_group_id,
            task_groups_list=task_group_list,
            db=db,
        )
        for wf_task in workflow_import.task_list
    ]

    if any(success is False for success, _ in list_results):
        raise HTTPExceptionWithData(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            data=[
                {
                    "outcome": "success",
                    "pkg_name": wf_task.task.pkg_name,
                    "version": wf_task.task.version,
                    "task_name": wf_task.task.name,
                    "task_id": task_id_or_available_tasks,
                }
                if success
                else {
                    "outcome": "fail",
                    "pkg_name": wf_task.task.pkg_name,
                    "version": wf_task.task.version,
                    "task_name": wf_task.task.name,
                    "available_tasks": [
                        available_task.model_dump()
                        for available_task in task_id_or_available_tasks
                    ],
                }
                for wf_task, (success, task_id_or_available_tasks) in zip(
                    workflow_import.task_list, list_results
                )
            ],
        )

    for wf_task, (_, task_id) in zip(workflow_import.task_list, list_results):
        new_wf_task = WorkflowTaskCreate(
            **wf_task.model_dump(exclude_none=True, exclude={"task"}),
            task_id=task_id,
        )
        list_wf_tasks.append(new_wf_task)
        task = await db.get(TaskV2, task_id)
        _check_type_filters_compatibility(
            task_input_types=task.input_types,
            wftask_type_filters=new_wf_task.type_filters,
        )

    # Create new Workflow
    db_workflow = WorkflowV2(
        project_id=project_id,
        template_id=template_id,
        **workflow_import.model_dump(exclude_none=True, exclude={"task_list"}),
    )
    db.add(db_workflow)
    await db.flush()
    await db.refresh(db_workflow)

    # Insert tasks into the workflow
    for i, (new_wf_task, (_, task_id)) in enumerate(
        zip(list_wf_tasks, list_results)
    ):
        await _workflow_insert_task(
            **new_wf_task.model_dump(),
            workflow_id=db_workflow.id,
            order=i,
            db=db,
        )

    # Add warnings for non-active tasks (or non-accessible tasks,
    # although that should never happen)
    wftask_list_with_warnings = await _add_warnings_to_workflow_tasks(
        wftask_list=db_workflow.task_list, user_id=user.id, db=db
    )
    workflow_data = dict(
        **db_workflow.model_dump(),
        project=db_workflow.project,
        task_list=wftask_list_with_warnings,
    )

    await db.commit()
    return workflow_data