Job

@router.get("/{job_id}/download/", response_class=StreamingResponse)
async def download_job_logs(
    job_id: int,
    user: UserOAuth = Depends(current_superuser_act),
    db: AsyncSession = Depends(get_async_db),
) -> StreamingResponse:
    """
    Download job folder
    """
    # Get job from DB
    job = await db.get(JobV2, job_id)
    if job is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Job {job_id} not found",
        )
    # Create and return byte stream for zipped log folder
    PREFIX_ZIP = Path(job.working_dir).name
    zip_filename = f"{PREFIX_ZIP}_archive.zip"
    return StreamingResponse(
        _zip_folder_to_byte_stream_iterator(folder=job.working_dir),
        media_type="application/x-zip-compressed",
        headers={"Content-Disposition": f"attachment;filename={zip_filename}"},
    )

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

    _check_shutdown_is_supported()

    job = await db.get(JobV2, job_id)
    if job is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Job {job_id} not found",
        )

    _raise_422_if_status_not_submitted(job=job)
    _write_shutdown_file_or_422(job=job)

    return Response(status_code=status.HTTP_202_ACCEPTED)

@router.patch("/{job_id}/", response_model=JobRead)
async def update_job(
    job_update: JobUpdate,
    job_id: int,
    user: UserOAuth = Depends(current_superuser_act),
    db: AsyncSession = Depends(get_async_db),
) -> JobV2:
    """
    Change the status of an existing job.

    This endpoint is only open to superusers, and it does not apply
    project-based access-control to jobs.
    """
    job = await db.get(JobV2, job_id)
    if job is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Job {job_id} not found",
        )
    if job.status != JobStatusType.SUBMITTED:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=f"Job {job_id} has status {job.status=} != 'submitted'.",
        )

    if job_update.status != JobStatusType.FAILED:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=f"Cannot set job status to {job_update.status}",
        )

    timestamp = get_timestamp()
    setattr(job, "status", job_update.status)
    setattr(job, "end_timestamp", timestamp)
    setattr(
        job,
        "log",
        f"{job.log or ''}\nThis job was manually marked as "
        f"'{JobStatusType.FAILED}' by an admin ({timestamp.isoformat()}).",
    )

    res = await db.execute(
        select(HistoryRun)
        .where(HistoryRun.job_id == job_id)
        .order_by(HistoryRun.timestamp_started.desc())
        .limit(1)
    )
    latest_run = res.scalar_one_or_none()
    if latest_run is not None:
        setattr(latest_run, "status", HistoryUnitStatus.FAILED)
        res = await db.execute(
            select(HistoryUnit).where(
                HistoryUnit.history_run_id == latest_run.id
            )
        )
        history_units = res.scalars().all()
        for history_unit in history_units:
            setattr(history_unit, "status", HistoryUnitStatus.FAILED)

    await db.commit()
    await db.refresh(job)
    return job

@router.get("/", response_model=PaginationResponse[JobRead])
async def view_job(
    id: int | None = None,
    resource_id: int | None = None,
    project_owner_id: int | None = None,
    job_user_email: EmailStr | None = None,
    project_id: int | None = None,
    dataset_id: int | None = None,
    workflow_id: int | None = None,
    status: JobStatusType | None = None,
    start_timestamp_min: AwareDatetime | None = None,
    start_timestamp_max: AwareDatetime | None = None,
    end_timestamp_min: AwareDatetime | None = None,
    end_timestamp_max: AwareDatetime | None = None,
    log: bool = True,
    pagination: PaginationRequest = Depends(get_pagination_params),
    user: UserOAuth = Depends(current_superuser_act),
    db: AsyncSession = Depends(get_async_db),
) -> PaginationResponse[JobV2]:
    """
    Query `JobV2` table.

    Args:
        id: If not `None`, select a given `applyworkflow.id`.
        project_owner_id:
        job_user_email:
        project_id: If not `None`, select a given `applyworkflow.project_id`.
        dataset_id: If not `None`, select a given
            `applyworkflow.input_dataset_id`.
        workflow_id: If not `None`, select a given `applyworkflow.workflow_id`.
        status: If not `None`, select a given `applyworkflow.status`.
        start_timestamp_min: If not `None`, select a rows with
            `start_timestamp` after `start_timestamp_min`.
        start_timestamp_max: If not `None`, select a rows with
            `start_timestamp` before `start_timestamp_min`.
        end_timestamp_min: If not `None`, select a rows with `end_timestamp`
            after `end_timestamp_min`.
        end_timestamp_max: If not `None`, select a rows with `end_timestamp`
            before `end_timestamp_min`.
        log: If `True`, include `job.log`, if `False`
            `job.log` is set to `None`.
    """

    # Prepare statements
    stm = select(JobV2).order_by(JobV2.start_timestamp.desc())
    stm_count = select(func.count(JobV2.id))
    if id is not None:
        stm = stm.where(JobV2.id == id)
        stm_count = stm_count.where(JobV2.id == id)
    if resource_id is not None:
        stm = stm.join(ProjectV2, ProjectV2.id == JobV2.project_id).where(
            ProjectV2.resource_id == resource_id
        )
        stm_count = stm_count.join(
            ProjectV2, ProjectV2.id == JobV2.project_id
        ).where(ProjectV2.resource_id == resource_id)
    if project_owner_id is not None:
        stm = (
            stm.join(
                LinkUserProjectV2,
                LinkUserProjectV2.project_id == JobV2.project_id,
            )
            .where(LinkUserProjectV2.user_id == project_owner_id)
            .where(LinkUserProjectV2.is_owner.is_(True))
        )
        stm_count = (
            stm_count.join(
                LinkUserProjectV2,
                LinkUserProjectV2.project_id == JobV2.project_id,
            )
            .where(LinkUserProjectV2.user_id == project_owner_id)
            .where(LinkUserProjectV2.is_owner.is_(True))
        )
    if job_user_email is not None:
        stm = stm.where(JobV2.user_email == job_user_email)
        stm_count = stm_count.where(JobV2.user_email == job_user_email)
    if project_id is not None:
        stm = stm.where(JobV2.project_id == project_id)
        stm_count = stm_count.where(JobV2.project_id == project_id)
    if dataset_id is not None:
        stm = stm.where(JobV2.dataset_id == dataset_id)
        stm_count = stm_count.where(JobV2.dataset_id == dataset_id)
    if workflow_id is not None:
        stm = stm.where(JobV2.workflow_id == workflow_id)
        stm_count = stm_count.where(JobV2.workflow_id == workflow_id)
    if status is not None:
        stm = stm.where(JobV2.status == status)
        stm_count = stm_count.where(JobV2.status == status)
    if start_timestamp_min is not None:
        stm = stm.where(JobV2.start_timestamp >= start_timestamp_min)
        stm_count = stm_count.where(
            JobV2.start_timestamp >= start_timestamp_min
        )
    if start_timestamp_max is not None:
        stm = stm.where(JobV2.start_timestamp <= start_timestamp_max)
        stm_count = stm_count.where(
            JobV2.start_timestamp <= start_timestamp_max
        )
    if end_timestamp_min is not None:
        stm = stm.where(JobV2.end_timestamp >= end_timestamp_min)
        stm_count = stm_count.where(JobV2.end_timestamp >= end_timestamp_min)
    if end_timestamp_max is not None:
        stm = stm.where(JobV2.end_timestamp <= end_timestamp_max)
        stm_count = stm_count.where(JobV2.end_timestamp <= end_timestamp_max)

    response = await get_paginated_response(
        stm=stm, stm_count=stm_count, pagination=pagination, db=db
    )

    if not log:
        for job in response.items:
            setattr(job, "log", None)

    return response