workflow

create_workflow(project_id, workflow, user=Depends(current_active_user), db=Depends(get_async_db)) async

Create a workflow, associate to a project

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

@router.post(
    "/project/{project_id}/workflow/",
    response_model=WorkflowReadV2,
    status_code=status.HTTP_201_CREATED,
)
async def create_workflow(
    project_id: int,
    workflow: WorkflowCreateV2,
    user: UserOAuth = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_db),
) -> Optional[WorkflowReadV2]:
    """
    Create a workflow, associate to a project
    """
    await _get_project_check_owner(
        project_id=project_id, user_id=user.id, db=db
    )
    await _check_workflow_exists(
        name=workflow.name, project_id=project_id, db=db
    )

    db_workflow = WorkflowV2(project_id=project_id, **workflow.model_dump())
    db.add(db_workflow)
    await db.commit()
    await db.refresh(db_workflow)
    await db.close()
    return db_workflow

delete_workflow(project_id, workflow_id, user=Depends(current_active_user), db=Depends(get_async_db)) async

Delete a workflow

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

@router.delete(
    "/project/{project_id}/workflow/{workflow_id}/",
    status_code=status.HTTP_204_NO_CONTENT,
)
async def delete_workflow(
    project_id: int,
    workflow_id: int,
    user: UserOAuth = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_db),
) -> Response:
    """
    Delete a workflow
    """

    workflow = await _get_workflow_check_owner(
        project_id=project_id,
        workflow_id=workflow_id,
        user_id=user.id,
        db=db,
    )

    # Fail if there exist jobs that are submitted and in relation with the
    # current workflow.
    stm = _get_submitted_jobs_statement().where(
        JobV2.workflow_id == workflow.id
    )
    res = await db.execute(stm)
    jobs = res.scalars().all()
    if jobs:
        string_ids = str([job.id for job in jobs])[1:-1]
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=(
                f"Cannot delete workflow {workflow.id} because it "
                f"is linked to active job(s) {string_ids}."
            ),
        )

    # Cascade operations: set foreign-keys to null for jobs which are in
    # relationship with the current workflow
    stm = select(JobV2).where(JobV2.workflow_id == workflow_id)
    res = await db.execute(stm)
    jobs = res.scalars().all()
    for job in jobs:
        job.workflow_id = None

    # Delete workflow
    await db.delete(workflow)
    await db.commit()

    return Response(status_code=status.HTTP_204_NO_CONTENT)

export_worfklow(project_id, workflow_id, user=Depends(current_active_user), db=Depends(get_async_db)) async

Export an existing workflow, after stripping all IDs

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

@router.get(
    "/project/{project_id}/workflow/{workflow_id}/export/",
    response_model=WorkflowExportV2,
)
async def export_worfklow(
    project_id: int,
    workflow_id: int,
    user: UserOAuth = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_db),
) -> Optional[WorkflowExportV2]:
    """
    Export an existing workflow, after stripping all IDs
    """
    workflow = await _get_workflow_check_owner(
        project_id=project_id,
        workflow_id=workflow_id,
        user_id=user.id,
        db=db,
    )
    wf_task_list = []
    for wftask in workflow.task_list:
        task_group = await db.get(TaskGroupV2, wftask.task.taskgroupv2_id)
        wf_task_list.append(wftask.model_dump())
        wf_task_list[-1]["task"] = dict(
            pkg_name=task_group.pkg_name,
            version=task_group.version,
            name=wftask.task.name,
        )

    wf = WorkflowExportV2(
        **workflow.model_dump(),
        task_list=wf_task_list,
    )
    return wf

get_user_workflows(user=Depends(current_active_user), db=Depends(get_async_db)) async

Returns all the workflows of the current user

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

@router.get("/workflow/", response_model=list[WorkflowReadV2])
async def get_user_workflows(
    user: UserOAuth = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_db),
) -> list[WorkflowReadV2]:
    """
    Returns all the workflows of the current user
    """
    stm = select(WorkflowV2)
    stm = stm.join(ProjectV2).where(
        ProjectV2.user_list.any(UserOAuth.id == user.id)
    )
    res = await db.execute(stm)
    workflow_list = res.scalars().all()
    return workflow_list

get_workflow_list(project_id, user=Depends(current_active_user), db=Depends(get_async_db)) async

Get workflow list for given project

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

@router.get(
    "/project/{project_id}/workflow/",
    response_model=list[WorkflowReadV2],
)
async def get_workflow_list(
    project_id: int,
    user: UserOAuth = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_db),
) -> Optional[list[WorkflowReadV2]]:
    """
    Get workflow list for given project
    """
    # Access control
    project = await _get_project_check_owner(
        project_id=project_id, user_id=user.id, db=db
    )
    # Find workflows of the current project. Note: this select/where approach
    # has much better scaling than refreshing all elements of
    # `project.workflow_list` - ref
    # https://github.com/fractal-analytics-platform/fractal-server/pull/1082#issuecomment-1856676097.
    stm = select(WorkflowV2).where(WorkflowV2.project_id == project.id)
    workflow_list = (await db.execute(stm)).scalars().all()
    return workflow_list

get_workflow_type_filters(project_id, workflow_id, dataset_id=None, first_task_index=None, last_task_index=None, user=Depends(current_active_user), db=Depends(get_async_db)) async

Get info on an existing workflow

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

@router.get(
    "/project/{project_id}/workflow/{workflow_id}/type-filters-flow/",
    response_model=TypeFiltersFlow,
)
async def get_workflow_type_filters(
    project_id: int,
    workflow_id: int,
    dataset_id: Optional[int] = None,
    first_task_index: Optional[int] = None,
    last_task_index: Optional[int] = None,
    user: UserOAuth = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_db),
) -> Optional[WorkflowReadV2WithWarnings]:
    """
    Get info on an existing workflow
    """

    workflow = await _get_workflow_check_owner(
        project_id=project_id,
        workflow_id=workflow_id,
        user_id=user.id,
        db=db,
    )

    if len(workflow.task_list) == 0:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail="Workflow has no tasks.",
        )

    if dataset_id is None:
        dataset_type_filters = {}
    else:
        res = await _get_dataset_check_owner(
            project_id=project_id,
            dataset_id=dataset_id,
            user_id=user.id,
            db=db,
        )
        dataset = res["dataset"]
        dataset_type_filters = dataset.type_filters

    num_tasks = len(workflow.task_list)
    try:
        first_task_index, last_task_index = set_start_and_last_task_index(
            num_tasks,
            first_task_index=first_task_index,
            last_task_index=last_task_index,
        )
    except ValueError as e:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"Invalid first/last task index.\nOriginal error: {str(e)}",
        )

    list_dataset_filters = [copy(dataset_type_filters)]
    list_filters_in = []
    list_filters_out = []
    for wftask in workflow.task_list[first_task_index : last_task_index + 1]:

        input_type_filters = copy(dataset_type_filters)
        patch = merge_type_filters(
            wftask_type_filters=wftask.type_filters,
            task_input_types=wftask.task.input_types,
        )
        input_type_filters.update(patch)
        list_filters_in.append(copy(input_type_filters))

        output_type_filters = wftask.task.output_types
        list_filters_out.append(output_type_filters)

        dataset_type_filters.update(wftask.task.output_types)
        list_dataset_filters.append(copy(dataset_type_filters))

    response_body = dict(
        dataset_filters=list_dataset_filters,
        input_filters=list_filters_in,
        output_filters=list_filters_out,
    )
    return response_body

read_workflow(project_id, workflow_id, user=Depends(current_active_user), db=Depends(get_async_db)) async

Get info on an existing workflow

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

@router.get(
    "/project/{project_id}/workflow/{workflow_id}/",
    response_model=WorkflowReadV2WithWarnings,
)
async def read_workflow(
    project_id: int,
    workflow_id: int,
    user: UserOAuth = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_db),
) -> Optional[WorkflowReadV2WithWarnings]:
    """
    Get info on an existing workflow
    """

    workflow = await _get_workflow_check_owner(
        project_id=project_id,
        workflow_id=workflow_id,
        user_id=user.id,
        db=db,
    )

    wftask_list_with_warnings = await _add_warnings_to_workflow_tasks(
        wftask_list=workflow.task_list, user_id=user.id, db=db
    )
    workflow_data = dict(
        **workflow.model_dump(),
        project=workflow.project,
        task_list=wftask_list_with_warnings,
    )

    return workflow_data

update_workflow(project_id, workflow_id, patch, user=Depends(current_active_user), db=Depends(get_async_db)) async

Edit a workflow

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

@router.patch(
    "/project/{project_id}/workflow/{workflow_id}/",
    response_model=WorkflowReadV2WithWarnings,
)
async def update_workflow(
    project_id: int,
    workflow_id: int,
    patch: WorkflowUpdateV2,
    user: UserOAuth = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_db),
) -> Optional[WorkflowReadV2WithWarnings]:
    """
    Edit a workflow
    """
    workflow = await _get_workflow_check_owner(
        project_id=project_id,
        workflow_id=workflow_id,
        user_id=user.id,
        db=db,
    )

    if patch.name:
        await _check_workflow_exists(
            name=patch.name, project_id=project_id, db=db
        )

    for key, value in patch.model_dump(exclude_unset=True).items():
        if key == "reordered_workflowtask_ids":
            current_workflowtask_ids = [
                wftask.id for wftask in workflow.task_list
            ]
            num_tasks = len(workflow.task_list)
            if len(value) != num_tasks or set(value) != set(
                current_workflowtask_ids
            ):
                raise HTTPException(
                    status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                    detail=(
                        "`reordered_workflowtask_ids` must be a permutation of"
                        f" {current_workflowtask_ids} (given {value})"
                    ),
                )
            for ind_wftask in range(num_tasks):
                new_order = value.index(workflow.task_list[ind_wftask].id)
                workflow.task_list[ind_wftask].order = new_order
        else:
            setattr(workflow, key, value)

    await db.commit()
    await db.refresh(workflow)
    await db.close()

    wftask_list_with_warnings = await _add_warnings_to_workflow_tasks(
        wftask_list=workflow.task_list, user_id=user.id, db=db
    )
    workflow_data = dict(
        **workflow.model_dump(),
        project=workflow.project,
        task_list=wftask_list_with_warnings,
    )

    return workflow_data