task

query_tasks(id=None, source=None, version=None, name=None, max_number_of_results=25, category=None, modality=None, author=None, user=Depends(current_active_superuser), db=Depends(get_async_db)) async

Query TaskV2 table and get informations about related items (WorkflowV2s and ProjectV2s)

Parameters:

Name	Type	Description	Default
id	Optional[int]	If not None, query for matching task.id.	None
source	Optional[str]	If not None, query for contained case insensitive task.source.	None
version	Optional[str]	If not None, query for matching task.version.	None
name	Optional[str]	If not None, query for contained case insensitive task.name.	None
max_number_of_results	int	The maximum length of the response.	25
category	Optional[str]		None
modality	Optional[str]		None
author	Optional[str]		None

Source code in fractal_server/app/routes/admin/v2/task.py

@router.get("/", response_model=list[TaskV2Info])
async def query_tasks(
    id: Optional[int] = None,
    source: Optional[str] = None,
    version: Optional[str] = None,
    name: Optional[str] = None,
    max_number_of_results: int = 25,
    category: Optional[str] = None,
    modality: Optional[str] = None,
    author: Optional[str] = None,
    user: UserOAuth = Depends(current_active_superuser),
    db: AsyncSession = Depends(get_async_db),
) -> list[TaskV2Info]:
    """
    Query `TaskV2` table and get informations about related items
    (WorkflowV2s and ProjectV2s)

    Args:
        id: If not `None`, query for matching `task.id`.
        source: If not `None`, query for contained case insensitive
            `task.source`.
        version: If not `None`, query for matching `task.version`.
        name: If not `None`, query for contained case insensitive `task.name`.
        max_number_of_results: The maximum length of the response.
        category:
        modality:
        author:
    """

    stm = select(TaskV2)

    if id is not None:
        stm = stm.where(TaskV2.id == id)
    if source is not None:
        stm = stm.where(TaskV2.source.icontains(source))
    if version is not None:
        stm = stm.where(TaskV2.version == version)
    if name is not None:
        stm = stm.where(TaskV2.name.icontains(name))
    if category is not None:
        stm = stm.where(func.lower(TaskV2.category) == category.lower())
    if modality is not None:
        stm = stm.where(func.lower(TaskV2.modality) == modality.lower())
    if author is not None:
        stm = stm.where(TaskV2.authors.icontains(author))

    res = await db.execute(stm)
    task_list = res.scalars().all()
    if len(task_list) > max_number_of_results:
        await db.close()
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=(
                f"Too many Tasks ({len(task_list)} > {max_number_of_results})."
                " Please add more query filters."
            ),
        )

    task_info_list = []

    for task in task_list:
        stm = (
            select(WorkflowV2)
            .join(WorkflowTaskV2)
            .where(WorkflowTaskV2.workflow_id == WorkflowV2.id)
            .where(WorkflowTaskV2.task_id == task.id)
        )
        res = await db.execute(stm)
        wf_list = res.scalars().all()

        task_info_list.append(
            dict(
                task=task.model_dump(),
                relationships=[
                    dict(
                        workflow_id=workflow.id,
                        workflow_name=workflow.name,
                        project_id=workflow.project.id,
                        project_name=workflow.project.name,
                        project_users=[
                            dict(id=user.id, email=user.email)
                            for user in workflow.project.user_list
                        ],
                    )
                    for workflow in wf_list
                ],
            )
        )

    return task_info_list