nopaque/app/namespaces/jobs.py

from flask import current_app, Flask
from flask_login import current_user
from flask_socketio import Namespace
from app import db, hashids, socketio
from app.decorators import socketio_admin_required, socketio_login_required
from app.models import Job, JobStatus


def _delete_job(app: Flask, job_id: int):
    with app.app_context():
        job = Job.query.get(job_id)
        job.delete()
        db.session.commit()


def _restart_job(app, job_id):
    with app.app_context():
        job = Job.query.get(job_id)
        job.restart()
        db.session.commit()


class UsersNamespace(Namespace):
    @socketio_login_required
    def on_delete(self, job_hashid: str) -> dict:
        job_id = hashids.decode(job_hashid)

        if not isinstance(job_id, int):
            return {'status': 400, 'statusText': 'Bad Request'}

        job = Job.query.get(job_id)

        if job is None:
            return {'status': 404, 'statusText': 'Not found'}

        if not (
            job.user == current_user
            or current_user.is_administrator
        ):
            return {'status': 403, 'statusText': 'Forbidden'}

        socketio.start_background_task(
            _delete_job,
            current_app._get_current_object(),
            job_id
        )

        return {
            'body': f'Job "{job.title}" marked for deletion',
            'status': 202,
            'statusText': 'Accepted'
        }

    @socketio_admin_required
    def on_log(self, job_hashid: str):
        job_id = hashids.decode(job_hashid)

        if not isinstance(job_id, int):
            return {'status': 400, 'statusText': 'Bad Request'}

        job = Job.query.get(job_id)

        if job is None:
            return {'status': 404, 'statusText': 'Not found'}

        if job.status not in [JobStatus.COMPLETED, JobStatus.FAILED]:
            return {'status': 409, 'statusText': 'Conflict'}

        with open(job.path / 'pipeline_data' / 'logs' / 'pyflow_log.txt') as log_file:
            log = log_file.read()

        return {
            'body': log,
            'status': 200,
            'statusText': 'Forbidden'
        }

    socketio_login_required
    def on_restart(self, job_hashid: str):
        job_id = hashids.decode(job_hashid)

        if not isinstance(job_id, int):
            return {'status': 400, 'statusText': 'Bad Request'}

        job = Job.query.get(job_id)

        if job is None:
            return {'status': 404, 'statusText': 'Not found'}

        if not (
            job.user == current_user
            or current_user.is_administrator
        ):
            return {'status': 403, 'statusText': 'Forbidden'}

        if job.status == JobStatus.FAILED:
            return {'status': 409, 'statusText': 'Conflict'}

        socketio.start_background_task(
            _restart_job,
            current_app._get_current_object(),
            job_id
        )

        return {
            'body': f'Job "{job.title}" marked for restarting',
            'status': 202,
            'statusText': 'Accepted'
        }