job

download_job_logs(project_id, job_id, user=Depends(get_api_guest), db=Depends(get_async_db)) async

Download zipped job folder

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

@router.get(
    "/project/{project_id}/job/{job_id}/download/",
    response_class=StreamingResponse,
)
async def download_job_logs(
    project_id: int,
    job_id: int,
    user: UserOAuth = Depends(get_api_guest),
    db: AsyncSession = Depends(get_async_db),
) -> StreamingResponse:
    """
    Download zipped job folder
    """
    output = await _get_job_check_access(
        project_id=project_id,
        job_id=job_id,
        user_id=user.id,
        required_permissions=ProjectPermissions.READ,
        db=db,
    )
    job = output["job"]
    zip_name = f"{Path(job.working_dir).name}_archive.zip"

    zip_bytes_iterator = await zip_folder_threaded(job.working_dir)

    return StreamingResponse(
        zip_bytes_iterator,
        media_type="application/x-zip-compressed",
        headers={"Content-Disposition": f"attachment;filename={zip_name}"},
    )

get_job_list(project_id, user=Depends(get_api_guest), log=True, db=Depends(get_async_db)) async

Get job list for given project

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

@router.get(
    "/project/{project_id}/job/",
    response_model=list[JobRead],
)
async def get_job_list(
    project_id: int,
    user: UserOAuth = Depends(get_api_guest),
    log: bool = True,
    db: AsyncSession = Depends(get_async_db),
) -> list[JobRead] | None:
    """
    Get job list for given project
    """
    project = await _get_project_check_access(
        project_id=project_id,
        user_id=user.id,
        required_permissions=ProjectPermissions.READ,
        db=db,
    )

    res = await db.execute(
        select(JobV2)
        .where(JobV2.project_id == project.id)
        .order_by(JobV2.start_timestamp.desc())
    )
    job_list = res.scalars().all()
    if not log:
        for job in job_list:
            setattr(job, "log", None)

    return job_list

get_user_jobs(user=Depends(get_api_guest), log=True, db=Depends(get_async_db)) async

Returns all the jobs from projects linked to the current user

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

@router.get("/job/", response_model=list[JobRead])
async def get_user_jobs(
    user: UserOAuth = Depends(get_api_guest),
    log: bool = True,
    db: AsyncSession = Depends(get_async_db),
) -> list[JobRead]:
    """
    Returns all the jobs from projects linked to the current user
    """
    stm = (
        select(JobV2)
        .join(
            LinkUserProjectV2, LinkUserProjectV2.project_id == JobV2.project_id
        )
        .where(LinkUserProjectV2.user_id == user.id)
        .where(LinkUserProjectV2.is_verified.is_(True))
        .order_by(JobV2.start_timestamp.desc())
    )
    res = await db.execute(stm)
    job_list = res.scalars().all()
    if not log:
        for job in job_list:
            setattr(job, "log", None)

    return job_list

get_workflow_jobs(project_id, workflow_id, user=Depends(get_api_guest), db=Depends(get_async_db)) async

Returns all the jobs related to a specific workflow

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

@router.get(
    "/project/{project_id}/workflow/{workflow_id}/job/",
    response_model=list[JobRead],
)
async def get_workflow_jobs(
    project_id: int,
    workflow_id: int,
    user: UserOAuth = Depends(get_api_guest),
    db: AsyncSession = Depends(get_async_db),
) -> list[JobRead] | None:
    """
    Returns all the jobs related to a specific workflow
    """
    await _get_workflow_check_access(
        project_id=project_id,
        workflow_id=workflow_id,
        user_id=user.id,
        required_permissions=ProjectPermissions.READ,
        db=db,
    )
    res = await db.execute(
        select(JobV2)
        .where(JobV2.workflow_id == workflow_id)
        .order_by(JobV2.start_timestamp.desc())
    )
    job_list = res.scalars().all()
    return job_list

read_job(project_id, job_id, show_tmp_logs=False, user=Depends(get_api_guest), db=Depends(get_async_db)) async

Return info on an existing job

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

@router.get(
    "/project/{project_id}/job/{job_id}/",
    response_model=JobRead,
)
async def read_job(
    project_id: int,
    job_id: int,
    show_tmp_logs: bool = False,
    user: UserOAuth = Depends(get_api_guest),
    db: AsyncSession = Depends(get_async_db),
) -> JobRead | None:
    """
    Return info on an existing job
    """

    output = await _get_job_check_access(
        project_id=project_id,
        job_id=job_id,
        user_id=user.id,
        required_permissions=ProjectPermissions.READ,
        db=db,
    )
    job = output["job"]

    if show_tmp_logs and (job.status == JobStatusType.SUBMITTED):
        try:
            with open(f"{job.working_dir}/{WORKFLOW_LOG_FILENAME}") as f:
                job.log = f.read()
        except FileNotFoundError:
            pass

    return job

stop_job(project_id, job_id, user=Depends(get_api_user), db=Depends(get_async_db)) async

Stop execution of a workflow job.

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

@router.get(
    "/project/{project_id}/job/{job_id}/stop/",
    status_code=202,
)
async def stop_job(
    project_id: int,
    job_id: int,
    user: UserOAuth = Depends(get_api_user),
    db: AsyncSession = Depends(get_async_db),
) -> Response:
    """
    Stop execution of a workflow job.
    """

    _check_shutdown_is_supported()

    # Get job from DB
    output = await _get_job_check_access(
        project_id=project_id,
        job_id=job_id,
        user_id=user.id,
        required_permissions=ProjectPermissions.EXECUTE,
        db=db,
    )
    job = output["job"]

    _write_shutdown_file(job=job)

    return Response(status_code=status.HTTP_202_ACCEPTED)