task_group

_version_sort_key(task_group)

Returns a tuple used as (reverse) ordering key for TaskGroups in get_task_group_list. The TaskGroups with a parsable versions are the first in order, sorted according to the sorting rules of packaging.version.Version. Next in order we have the TaskGroups with non-null non-parsable versions, sorted alphabetically. Last we have the TaskGroups with null version.

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

def _version_sort_key(
    task_group: TaskGroupV2,
) -> tuple[int, Version | str | None]:
    """
    Returns a tuple used as (reverse) ordering key for TaskGroups in
    `get_task_group_list`.
    The TaskGroups with a parsable versions are the first in order,
    sorted according to the sorting rules of packaging.version.Version.
    Next in order we have the TaskGroups with non-null non-parsable versions,
    sorted alphabetically.
    Last we have the TaskGroups with null version.
    """
    if task_group.version is None:
        return (0, task_group.version)
    try:
        return (2, parse(task_group.version))
    except InvalidVersion:
        return (1, task_group.version)

get_task_group(task_group_id, user=Depends(current_active_user), db=Depends(get_async_db)) async

Get single TaskGroup

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

@router.get("/{task_group_id}/", response_model=TaskGroupReadV2)
async def get_task_group(
    task_group_id: int,
    user: UserOAuth = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_db),
) -> TaskGroupReadV2:
    """
    Get single TaskGroup
    """
    task_group = await _get_task_group_read_access(
        task_group_id=task_group_id,
        user_id=user.id,
        db=db,
    )
    return task_group

get_task_group_list(user=Depends(current_active_user), db=Depends(get_async_db), only_active=False, only_owner=False, args_schema=True) async

Get all accessible TaskGroups

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

@router.get("/", response_model=list[tuple[str, list[TaskGroupReadV2]]])
async def get_task_group_list(
    user: UserOAuth = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_db),
    only_active: bool = False,
    only_owner: bool = False,
    args_schema: bool = True,
) -> list[tuple[str, list[TaskGroupReadV2]]]:
    """
    Get all accessible TaskGroups
    """
    if only_owner:
        condition = TaskGroupV2.user_id == user.id
    else:
        condition = or_(
            TaskGroupV2.user_id == user.id,
            TaskGroupV2.user_group_id.in_(
                select(LinkUserGroup.group_id).where(
                    LinkUserGroup.user_id == user.id
                )
            ),
        )
    stm = select(TaskGroupV2).where(condition).order_by(TaskGroupV2.pkg_name)
    if only_active:
        stm = stm.where(TaskGroupV2.active)

    res = await db.execute(stm)
    task_groups = res.scalars().all()

    if args_schema is False:
        for taskgroup in task_groups:
            for task in taskgroup.task_list:
                setattr(task, "args_schema_non_parallel", None)
                setattr(task, "args_schema_parallel", None)

    default_group_id = await _get_default_usergroup_id(db)
    grouped_result = [
        (
            pkg_name,
            (
                await remove_duplicate_task_groups(
                    task_groups=sorted(
                        list(groups),
                        key=_version_sort_key,
                        reverse=True,
                    ),
                    user_id=user.id,
                    default_group_id=default_group_id,
                    db=db,
                )
            ),
        )
        for pkg_name, groups in itertools.groupby(
            task_groups, key=lambda tg: tg.pkg_name
        )
    ]
    return grouped_result

patch_task_group(task_group_id, task_group_update, user=Depends(current_active_user), db=Depends(get_async_db)) async

Patch single TaskGroup

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

@router.patch("/{task_group_id}/", response_model=TaskGroupReadV2)
async def patch_task_group(
    task_group_id: int,
    task_group_update: TaskGroupUpdateV2,
    user: UserOAuth = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_db),
) -> TaskGroupReadV2:
    """
    Patch single TaskGroup
    """
    task_group = await _get_task_group_full_access(
        task_group_id=task_group_id,
        user_id=user.id,
        db=db,
    )
    if (
        "user_group_id" in task_group_update.model_dump(exclude_unset=True)
        and task_group_update.user_group_id != task_group.user_group_id
    ):
        await _verify_non_duplication_group_constraint(
            db=db,
            pkg_name=task_group.pkg_name,
            version=task_group.version,
            user_group_id=task_group_update.user_group_id,
        )
    for key, value in task_group_update.model_dump(exclude_unset=True).items():
        if (key == "user_group_id") and (value is not None):
            await _verify_user_belongs_to_group(
                user_id=user.id, user_group_id=value, db=db
            )
        setattr(task_group, key, value)

    db.add(task_group)
    await db.commit()
    await db.refresh(task_group)
    return task_group