task_group

get_task_group(task_group_id, user=Depends(get_api_guest), 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=TaskGroupRead)
async def get_task_group(
    task_group_id: int,
    user: UserOAuth = Depends(get_api_guest),
    db: AsyncSession = Depends(get_async_db),
) -> TaskGroupRead:
    """
    Get single TaskGroup
    """
    task_group = await _get_task_group_read_access(
        task_group_id=task_group_id,
        user_id=user.id,
        db=db,
    )
    task_group_with_email = await add_user_email_to_task_group(
        task_group=task_group, db=db
    )
    return task_group_with_email

get_task_group_list(user=Depends(get_api_guest), 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[TaskGroupRead]]])
async def get_task_group_list(
    user: UserOAuth = Depends(get_api_guest),
    db: AsyncSession = Depends(get_async_db),
    only_active: bool = False,
    only_owner: bool = False,
    args_schema: bool = True,
) -> list[tuple[str, list[TaskGroupRead]]]:
    """
    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
                )
            ),
        )

    user_resource_id = await _get_user_resource_id(user_id=user.id, db=db)
    stm = (
        select(TaskGroupV2, UserOAuth.email)
        .join(UserOAuth, UserOAuth.id == TaskGroupV2.user_id)
        .where(TaskGroupV2.resource_id == user_resource_id)
        .where(condition)
        .order_by(TaskGroupV2.pkg_name)
    )
    if only_active:
        stm = stm.where(TaskGroupV2.active)

    res = await db.execute(stm)
    task_groups_and_email = res.all()

    task_groups = [item[0] for item in task_groups_and_email]
    task_group_id_email_map = {
        task_group.id: user_email
        for task_group, user_email in task_groups_and_email
    }

    if args_schema is False:
        for taskgroup in task_groups:
            for task in taskgroup.task_list:
                db.expunge(task)  # See issue 3101
                setattr(task, "args_schema_non_parallel", None)
                setattr(task, "args_schema_parallel", None)

    default_group_id = await _get_default_usergroup_id_or_none(db)
    grouped_result = [
        (
            pkg_name,
            (
                await remove_duplicate_task_groups(
                    task_groups=sorted(
                        list(groups),
                        key=lambda group: _version_sort_key(group.version),
                        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
        )
    ]
    grouped_result_with_emails = [
        (
            pkg_name,
            [
                serialize_task_group_with_email(
                    task_group=task_group,
                    user_email=task_group_id_email_map[task_group.id],
                )
                for task_group in task_group_list
            ],
        )
        for pkg_name, task_group_list in grouped_result
    ]

    return grouped_result_with_emails

patch_task_group(task_group_id, task_group_update, user=Depends(get_api_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=TaskGroupRead)
async def patch_task_group(
    task_group_id: int,
    task_group_update: TaskGroupUpdate,
    user: UserOAuth = Depends(get_api_user),
    db: AsyncSession = Depends(get_async_db),
) -> TaskGroupRead:
    """
    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)
    task_group_with_email = await add_user_email_to_task_group(
        task_group=task_group, db=db
    )
    return task_group_with_email