workflowtask

create_workflowtask(project_id, workflow_id, task_id, new_task, user=Depends(current_active_user), db=Depends(get_async_db)) async

Add a WorkflowTask to a Workflow

Source code in fractal_server/app/routes/api/v1/workflowtask.py

@router.post(
    "/project/{project_id}/workflow/{workflow_id}/wftask/",
    response_model=WorkflowTaskReadV1,
    status_code=status.HTTP_201_CREATED,
)
async def create_workflowtask(
    project_id: int,
    workflow_id: int,
    task_id: int,
    new_task: WorkflowTaskCreateV1,
    user: UserOAuth = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_db),
) -> Optional[WorkflowTaskReadV1]:
    """
    Add a WorkflowTask to a Workflow
    """
    _raise_if_v1_is_read_only()
    workflow = await _get_workflow_check_owner(
        project_id=project_id, workflow_id=workflow_id, user_id=user.id, db=db
    )

    # Check that task exists
    task = await db.get(Task, task_id)
    if not task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Task {task_id} not found.",
        )

    async with db:
        workflow_task = await _workflow_insert_task(
            **new_task.dict(),
            workflow_id=workflow.id,
            task_id=task_id,
            db=db,
        )

    await db.close()
    return workflow_task

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

Delete a WorkflowTask of a Workflow

Source code in fractal_server/app/routes/api/v1/workflowtask.py

@router.delete(
    "/project/{project_id}/workflow/{workflow_id}/wftask/{workflow_task_id}/",
    status_code=status.HTTP_204_NO_CONTENT,
)
async def delete_workflowtask(
    project_id: int,
    workflow_id: int,
    workflow_task_id: int,
    user: UserOAuth = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_db),
) -> Response:
    """
    Delete a WorkflowTask of a Workflow
    """
    _raise_if_v1_is_read_only()
    db_workflow_task, db_workflow = await _get_workflow_task_check_owner(
        project_id=project_id,
        workflow_task_id=workflow_task_id,
        workflow_id=workflow_id,
        user_id=user.id,
        db=db,
    )

    await db.delete(db_workflow_task)
    await db.commit()

    await db.refresh(db_workflow)
    db_workflow.task_list.reorder()
    await db.commit()

    return Response(status_code=status.HTTP_204_NO_CONTENT)

update_workflowtask(project_id, workflow_id, workflow_task_id, workflow_task_update, user=Depends(current_active_user), db=Depends(get_async_db)) async

Edit a WorkflowTask of a Workflow

Source code in fractal_server/app/routes/api/v1/workflowtask.py

@router.patch(
    "/project/{project_id}/workflow/{workflow_id}/wftask/{workflow_task_id}/",
    response_model=WorkflowTaskReadV1,
)
async def update_workflowtask(
    project_id: int,
    workflow_id: int,
    workflow_task_id: int,
    workflow_task_update: WorkflowTaskUpdateV1,
    user: UserOAuth = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_db),
) -> Optional[WorkflowTaskReadV1]:
    """
    Edit a WorkflowTask of a Workflow
    """
    _raise_if_v1_is_read_only()
    db_workflow_task, db_workflow = await _get_workflow_task_check_owner(
        project_id=project_id,
        workflow_task_id=workflow_task_id,
        workflow_id=workflow_id,
        user_id=user.id,
        db=db,
    )

    for key, value in workflow_task_update.dict(exclude_unset=True).items():
        if key == "args":

            # Get default arguments via a Task property method
            default_args = deepcopy(
                db_workflow_task.task.default_args_from_args_schema
            )
            # Override default_args with args value items
            actual_args = default_args.copy()
            if value is not None:
                for k, v in value.items():
                    actual_args[k] = v
            if not actual_args:
                actual_args = None
            setattr(db_workflow_task, key, actual_args)
        elif key == "meta":
            current_meta = deepcopy(db_workflow_task.meta) or {}
            current_meta.update(value)
            setattr(db_workflow_task, key, current_meta)
        else:
            raise HTTPException(
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                detail=f"patch_workflow_task endpoint cannot set {key=}",
            )

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

    return db_workflow_task