task

create_task(task, user=Depends(current_active_verified_user), db=Depends(get_async_db)) async

Create a new task

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

@router.post(
    "/", response_model=TaskReadV1, status_code=status.HTTP_201_CREATED
)
async def create_task(
    task: TaskCreateV1,
    user: UserOAuth = Depends(current_active_verified_user),
    db: AsyncSession = Depends(get_async_db),
) -> Optional[TaskReadV1]:
    """
    Create a new task
    """
    _raise_if_v1_is_read_only()
    # Set task.owner attribute
    if user.username:
        owner = user.username
    else:
        verify_user_has_settings(user)
        if user.settings.slurm_user:
            owner = user.settings.slurm_user
        else:
            raise HTTPException(
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                detail=(
                    "Cannot add a new task because current user does not "
                    "have `username` or `slurm_user` attributes."
                ),
            )

    # Prepend owner to task.source
    task.source = f"{owner}:{task.source}"

    # Verify that source is not already in use (note: this check is only useful
    # to provide a user-friendly error message, but `task.source` uniqueness is
    # already guaranteed by a constraint in the table definition).
    stm = select(Task).where(Task.source == task.source)
    res = await db.execute(stm)
    if res.scalars().all():
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"Source '{task.source}' already used by some TaskV1",
        )
    stm = select(TaskV2).where(TaskV2.source == task.source)
    res = await db.execute(stm)
    if res.scalars().all():
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"Source '{task.source}' already used by some TaskV2",
        )

    # Add task
    db_task = Task(**task.dict(), owner=owner)
    db.add(db_task)
    await db.commit()
    await db.refresh(db_task)
    await db.close()
    return db_task

delete_task(task_id, user=Depends(current_active_user), db=Depends(get_async_db)) async

Delete a task

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

@router.delete("/{task_id}/", status_code=204)
async def delete_task(
    task_id: int,
    user: UserOAuth = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_db),
) -> Response:
    """
    Delete a task
    """
    _raise_if_v1_is_read_only()
    db_task = await _get_task_check_owner(task_id=task_id, user=user, db=db)

    # Check that the Task is not in relationship with some WorkflowTask
    stm = select(WorkflowTask).filter(WorkflowTask.task_id == task_id)
    res = await db.execute(stm)
    workflowtask_list = res.scalars().all()
    if workflowtask_list:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=(
                f"Cannot remove Task {task_id} because it is currently "
                "imported in Workflows "
                f"{[x.workflow_id for x in workflowtask_list]}. "
                "If you want to remove this task, then you should first remove"
                " the workflows.",
            ),
        )

    await db.delete(db_task)
    await db.commit()
    return Response(status_code=status.HTTP_204_NO_CONTENT)

get_list_task(user=Depends(current_active_user), args_schema=True, db=Depends(get_async_db)) async

Get list of available tasks

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

@router.get("/", response_model=list[TaskReadV1])
async def get_list_task(
    user: UserOAuth = Depends(current_active_user),
    args_schema: bool = True,
    db: AsyncSession = Depends(get_async_db),
) -> list[TaskReadV1]:
    """
    Get list of available tasks
    """
    stm = select(Task)
    res = await db.execute(stm)
    task_list = res.scalars().all()
    await db.close()
    if not args_schema:
        for task in task_list:
            setattr(task, "args_schema", None)

    return task_list

get_task(task_id, user=Depends(current_active_user), db=Depends(get_async_db)) async

Get info on a specific task

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

@router.get("/{task_id}/", response_model=TaskReadV1)
async def get_task(
    task_id: int,
    user: UserOAuth = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_db),
) -> TaskReadV1:
    """
    Get info on a specific task
    """
    task = await db.get(Task, task_id)
    await db.close()
    if not task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Task not found"
        )
    return task

patch_task(task_id, task_update, user=Depends(current_active_verified_user), db=Depends(get_async_db)) async

Edit a specific task (restricted to superusers and task owner)

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

@router.patch("/{task_id}/", response_model=TaskReadV1)
async def patch_task(
    task_id: int,
    task_update: TaskUpdateV1,
    user: UserOAuth = Depends(current_active_verified_user),
    db: AsyncSession = Depends(get_async_db),
) -> Optional[TaskReadV1]:
    """
    Edit a specific task (restricted to superusers and task owner)
    """
    _raise_if_v1_is_read_only()
    if task_update.source:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail="patch_task endpoint cannot set `source`",
        )

    # Retrieve task from database
    db_task = await _get_task_check_owner(task_id=task_id, user=user, db=db)

    update = task_update.dict(exclude_unset=True)
    for key, value in update.items():
        if isinstance(value, str) or (
            key == "version" and value is None
        ):  # special case (issue 817)
            setattr(db_task, key, value)
        elif isinstance(value, dict):
            if key == "args_schema":
                setattr(db_task, key, value)
            else:
                current_dict = deepcopy(getattr(db_task, key))
                current_dict.update(value)
                setattr(db_task, key, current_dict)
        else:
            raise HTTPException(
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                detail=f"Invalid {type(value)=} for {key=}",
            )

    await db.commit()
    await db.refresh(db_task)
    await db.close()
    return db_task