Workflow

@router.post(
    "/project/{project_id}/workflow/",
    response_model=WorkflowRead,
    status_code=status.HTTP_201_CREATED,
)
async def create_workflow(
    project_id: int,
    workflow: WorkflowCreate,
    user: UserOAuth = Depends(get_api_user),
    db: AsyncSession = Depends(get_async_db),
) -> WorkflowV2:
    """
    Create a workflow, associate to a project
    """
    await _get_project_check_access(
        project_id=project_id,
        user_id=user.id,
        required_permissions=ProjectPermissions.WRITE,
        db=db,
    )
    await _check_workflow_exists(
        name=workflow.name, project_id=project_id, db=db
    )

    db_workflow = WorkflowV2(project_id=project_id, **workflow.model_dump())
    db.add(db_workflow)
    await db.commit()
    await db.refresh(db_workflow)
    return db_workflow

@router.delete(
    "/project/{project_id}/workflow/{workflow_id}/",
    status_code=status.HTTP_204_NO_CONTENT,
)
async def delete_workflow(
    project_id: int,
    workflow_id: int,
    user: UserOAuth = Depends(get_api_user),
    db: AsyncSession = Depends(get_async_db),
) -> Response:
    """
    Delete a workflow
    """

    workflow = await _get_workflow_check_access(
        project_id=project_id,
        workflow_id=workflow_id,
        user_id=user.id,
        required_permissions=ProjectPermissions.WRITE,
        db=db,
    )

    # Fail if there exist jobs that are submitted and in relation with the
    # current workflow.
    stm = _get_submitted_jobs_statement().where(
        JobV2.workflow_id == workflow.id
    )
    res = await db.execute(stm)
    jobs = res.scalars().all()
    if jobs:
        string_ids = str([job.id for job in jobs])[1:-1]
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=(
                f"Cannot delete workflow {workflow.id} because it "
                f"is linked to active job(s) {string_ids}."
            ),
        )

    # Delete workflow
    await db.delete(workflow)
    await db.commit()

    return Response(status_code=status.HTTP_204_NO_CONTENT)

@router.get(
    "/project/{project_id}/workflow/{workflow_id}/export/",
    response_model=WorkflowExport,
)
async def export_workflow(
    project_id: int,
    workflow_id: int,
    user: UserOAuth = Depends(get_api_guest),
    db: AsyncSession = Depends(get_async_db),
) -> WorkflowExport:
    """
    Export an existing workflow, after stripping all IDs
    """
    workflow = await _get_workflow_check_access(
        project_id=project_id,
        workflow_id=workflow_id,
        user_id=user.id,
        required_permissions=ProjectPermissions.READ,
        db=db,
    )
    workflow_export = await _create_workflow_export(workflow=workflow, db=db)
    return workflow_export

@router.get(
    "/project/{project_id}/workflow/",
    response_model=list[WorkflowRead],
)
async def get_workflow_list(
    project_id: int,
    user: UserOAuth = Depends(get_api_guest),
    db: AsyncSession = Depends(get_async_db),
) -> Sequence[WorkflowV2]:
    """
    Get workflow list for given project
    """
    # Access control
    project = await _get_project_check_access(
        project_id=project_id,
        user_id=user.id,
        required_permissions=ProjectPermissions.READ,
        db=db,
    )
    # Find workflows of the current project. Note: this select/where approach
    # has much better scaling than refreshing all elements of
    # `project.workflow_list` - ref
    # https://github.com/fractal-analytics-platform/fractal-server/pull/1082#issuecomment-1856676097.
    res = await db.execute(
        select(WorkflowV2)
        .where(WorkflowV2.project_id == project.id)
        .order_by(
            WorkflowV2.is_starred.desc(),
            WorkflowV2.name,
        )
    )
    workflow_list = res.scalars().all()
    return workflow_list

@router.get("/project/{project_id}/workflow/{workflow_id}/type-filters-flow/")
async def get_workflow_type_filters(
    project_id: int,
    workflow_id: int,
    user: UserOAuth = Depends(get_api_guest),
    db: AsyncSession = Depends(get_async_db),
) -> list[dict[str, Any]]:
    """
    Get info on type/type-filters flow for a workflow.
    """

    workflow = await _get_workflow_check_access(
        project_id=project_id,
        workflow_id=workflow_id,
        user_id=user.id,
        required_permissions=ProjectPermissions.READ,
        db=db,
    )

    num_tasks = len(workflow.task_list)
    if num_tasks == 0:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail="Workflow has no tasks.",
        )

    current_type_filters = {}

    response_items = []
    for wftask in workflow.task_list:
        # Compute input_type_filters, based on wftask and task manifest
        input_type_filters = merge_type_filters(
            wftask_type_filters=wftask.type_filters,
            task_input_types=wftask.task.input_types,
        )

        # Append current item to response list
        response_items.append(
            dict(
                workflowtask_id=wftask.id,
                current_type_filters=copy(current_type_filters),
                input_type_filters=copy(input_type_filters),
                output_type_filters=copy(wftask.task.output_types),
            )
        )

        # Update `current_type_filters`
        current_type_filters.update(wftask.task.output_types)

    return response_items

@router.get(
    "/project/{project_id}/workflow/{workflow_id}/",
    response_model=WorkflowReadWithWarnings,
)
async def read_workflow(
    project_id: int,
    workflow_id: int,
    user: UserOAuth = Depends(get_api_guest),
    db: AsyncSession = Depends(get_async_db),
) -> dict[str, Any]:
    """
    Get info on an existing workflow
    """

    workflow = await _get_workflow_check_access(
        project_id=project_id,
        workflow_id=workflow_id,
        user_id=user.id,
        required_permissions=ProjectPermissions.READ,
        db=db,
    )

    wftask_list_with_warnings = await _add_warnings_to_workflow_tasks(
        wftask_list=workflow.task_list, user_id=user.id, db=db
    )
    workflow_data = dict(
        **workflow.model_dump(),
        project=workflow.project,
        task_list=wftask_list_with_warnings,
    )

    return workflow_data

@router.post(
    "/project/{project_id}/workflow/{workflow_id}/star/",
    status_code=status.HTTP_200_OK,
)
async def star_workflow(
    project_id: int,
    workflow_id: int,
    user: UserOAuth = Depends(get_api_user),
    db: AsyncSession = Depends(get_async_db),
) -> Response:
    """
    Set `WorkflowV2.is_starred` to `True`
    """
    workflow = await _get_workflow_check_access(
        project_id=project_id,
        workflow_id=workflow_id,
        user_id=user.id,
        required_permissions=ProjectPermissions.WRITE,
        db=db,
    )
    workflow.is_starred = True
    db.add(workflow)
    await db.commit()
    return Response(status_code=status.HTTP_200_OK)

@router.post(
    "/project/{project_id}/workflow/{workflow_id}/unstar/",
    status_code=status.HTTP_200_OK,
)
async def unstar_workflow(
    project_id: int,
    workflow_id: int,
    user: UserOAuth = Depends(get_api_user),
    db: AsyncSession = Depends(get_async_db),
) -> Response:
    """
    Set `WorkflowV2.is_starred` to `False`
    """
    workflow = await _get_workflow_check_access(
        project_id=project_id,
        workflow_id=workflow_id,
        user_id=user.id,
        required_permissions=ProjectPermissions.WRITE,
        db=db,
    )
    workflow.is_starred = False
    db.add(workflow)
    await db.commit()
    return Response(status_code=status.HTTP_200_OK)

@router.patch(
    "/project/{project_id}/workflow/{workflow_id}/",
    response_model=WorkflowReadWithWarnings,
)
async def update_workflow(
    project_id: int,
    workflow_id: int,
    patch: WorkflowUpdate,
    user: UserOAuth = Depends(get_api_user),
    db: AsyncSession = Depends(get_async_db),
) -> dict[str, Any]:
    """
    Edit a workflow
    """
    workflow = await _get_workflow_check_access(
        project_id=project_id,
        workflow_id=workflow_id,
        user_id=user.id,
        required_permissions=ProjectPermissions.WRITE,
        db=db,
    )

    if patch.name and patch.name != workflow.name:
        await _check_workflow_exists(
            name=patch.name, project_id=project_id, db=db
        )

    for key, value in patch.model_dump(exclude_unset=True).items():
        if key == "reordered_workflowtask_ids":
            if await _workflow_has_submitted_job(
                workflow_id=workflow_id, db=db
            ):
                raise HTTPException(
                    status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
                    detail=(
                        "Cannot re-order WorkflowTasks while a Job is running "
                        "for this Workflow."
                    ),
                )

            current_workflowtask_ids = [
                wftask.id for wftask in workflow.task_list
            ]
            num_tasks = len(workflow.task_list)
            if len(value) != num_tasks or set(value) != set(
                current_workflowtask_ids
            ):
                raise HTTPException(
                    status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
                    detail=(
                        "`reordered_workflowtask_ids` must be a permutation of"
                        f" {current_workflowtask_ids} (given {value})"
                    ),
                )
            for ind_wftask in range(num_tasks):
                new_order = value.index(workflow.task_list[ind_wftask].id)
                workflow.task_list[ind_wftask].order = new_order
        else:
            setattr(workflow, key, value)

    await db.commit()
    await db.refresh(workflow)

    wftask_list_with_warnings = await _add_warnings_to_workflow_tasks(
        wftask_list=workflow.task_list, user_id=user.id, db=db
    )
    workflow_data = dict(
        **workflow.model_dump(),
        project=workflow.project,
        task_list=wftask_list_with_warnings,
    )

    return workflow_data