new event system first draft

app/admin/events.py

@@ -0,0 +1,49 @@
+from flask_login import current_user
+from flask_socketio import disconnect, Namespace
+from app import db, hashids
+from app.extensions.flask_socketio_extras import admin_required
+from app.models import User
+
+
+class AdminNamespace(Namespace):
+    def on_connect(self):
+        # Check if the user is authenticated and is an administrator
+        if not (current_user.is_authenticated and current_user.is_administrator):
+            disconnect()
+
+
+    @admin_required
+    def on_set_user_confirmed(self, user_hashid: str, confirmed_value: bool):
+        # Decode the user hashid
+        user_id = hashids.decode(user_hashid)
+
+        # Validate user_id
+        if not isinstance(user_id, int):
+            return {
+                'code': 400,
+                'body': 'user_id is invalid'
+            }
+
+        # Validate confirmed_value
+        if not isinstance(confirmed_value, bool):
+            return {
+                'code': 400,
+                'body': 'confirmed_value is invalid'
+            }
+
+        # Load user from database
+        user = User.query.get(user_id)
+        if user is None:
+            return {
+                'code': 404,
+                'body': 'User not found'
+            }
+
+        # Update user confirmed status
+        user.confirmed = confirmed_value
+        db.session.commit()
+
+        return {
+            'code': 200,
+            'body': f'User "{user.username}" is now {"confirmed" if confirmed_value else "unconfirmed"}'
+        }

app/corpora/cqi_over_sio/__init__.py

@@ -9,7 +9,7 @@ from inspect import signature
 from threading import Lock
 from typing import Callable, Dict, List, Optional
 from app import db, docker_client, hashids, socketio
-from app.decorators import socketio_login_required
+from app.extensions.flask_socketio_extras import login_required
 from app.models import Corpus, CorpusStatus
 from . import extensions
 
@@ -87,11 +87,11 @@ CQI_API_FUNCTION_NAMES: List[str] = [
 
 
 class CQiNamespace(Namespace):
-    @socketio_login_required
+    @login_required
     def on_connect(self):
         pass
 
-    @socketio_login_required
+    @login_required
     def on_init(self, db_corpus_hashid: str):
         db_corpus_id: int = hashids.decode(db_corpus_hashid)
         db_corpus: Optional[Corpus] = Corpus.query.get(db_corpus_id)
@@ -134,7 +134,7 @@ class CQiNamespace(Namespace):
         }
         return {'code': 200, 'msg': 'OK'}
 
-    @socketio_login_required
+    @login_required
     def on_exec(self, fn_name: str, fn_args: Dict = {}):
         try:
             cqi_client: CQiClient = session['cqi_over_sio']['cqi_client']

app/corpora/events.py

@@ -1,12 +1,12 @@
 from flask_login import current_user
 from flask_socketio import join_room
 from app import hashids, socketio
-from app.decorators import socketio_login_required
+from app.extensions.flask_socketio_extras import login_required
 from app.models import Corpus
 
 
 @socketio.on('GET /corpora/<corpus_id>')
-@socketio_login_required
+@login_required
 def get_corpus(corpus_hashid):
     corpus_id = hashids.decode(corpus_hashid)
     corpus = Corpus.query.get(corpus_id)
@@ -29,7 +29,7 @@ def get_corpus(corpus_hashid):
 
 
 @socketio.on('SUBSCRIBE /corpora/<corpus_id>')
-@socketio_login_required
+@login_required
 def subscribe_corpus(corpus_hashid):
     corpus_id = hashids.decode(corpus_hashid)
     corpus = Corpus.query.get(corpus_id)

app/decorators.py

@@ -22,31 +22,6 @@ def admin_required(f):
     return permission_required(Permission.ADMINISTRATE)(f)
 
 
-def socketio_login_required(f):
-    @wraps(f)
-    def decorated_function(*args, **kwargs):
-        if current_user.is_authenticated:
-            return f(*args, **kwargs)
-        else:
-            return {'code': 401, 'msg': 'Unauthorized'}
-    return decorated_function
-
-
-def socketio_permission_required(permission):
-    def decorator(f):
-        @wraps(f)
-        def decorated_function(*args, **kwargs):
-            if not current_user.can(permission):
-                return {'code': 403, 'msg': 'Forbidden'}
-            return f(*args, **kwargs)
-        return decorated_function
-    return decorator
-
-
-def socketio_admin_required(f):
-    return socketio_permission_required(Permission.ADMINISTRATE)(f)
-
-
 def background(f):
     '''
     ' This decorator executes a function in a Thread.

app/extensions/flask_socketio_extras/__init__.py

@@ -0,0 +1,3 @@
+from .decorators import login_required
+from .decorators import permission_required
+from .decorators import admin_required

app/extensions/flask_socketio_extras/decorators.py

@@ -0,0 +1,27 @@
+from flask_login import current_user
+from functools import wraps
+from app.models import Permission as UserPermission
+
+
+def login_required(f):
+    @wraps(f)
+    def wrapper(*args, **kwargs):
+        if current_user.is_authenticated:
+            return f(*args, **kwargs)
+        return {'code': 401, 'body': 'Unauthorized'}
+    return wrapper
+
+
+def permission_required(permission):
+    def decorator(f):
+        @wraps(f)
+        def wrapper(*args, **kwargs):
+            if not current_user.can(permission):
+                return {'code': 403, 'body': 'Forbidden'}
+            return f(*args, **kwargs)
+        return wrapper
+    return decorator
+
+
+def admin_required(f):
+    return permission_required(UserPermission.ADMINISTRATE)(f)

app/jobs/events.py

@@ -0,0 +1,138 @@
+from flask import current_app
+from flask_login import current_user
+from flask_socketio import Namespace
+from app import db, hashids, socketio
+from app.extensions.flask_socketio import admin_required, login_required
+from app.models import Job, JobStatus
+
+
+class JobsNamespace(Namespace):
+    @login_required
+    def on_delete(self, job_hashid: str):
+        # Decode the job hashid
+        job_id = hashids.decode(job_hashid)
+
+        # Validate job_id
+        if not isinstance(job_id, int):
+            return {
+                'code': 400,
+                'body': 'job_id is invalid'
+            }
+
+        # Load job from database
+        job = Job.query.get(job_id)
+        if job is None:
+            return {
+                'code': 404,
+                'body': 'Job not found'
+            }
+
+        # Check if the current user is allowed to delete the job
+        if not (job.user == current_user or current_user.is_administrator):
+            return {
+                'code': 403,
+                'body': 'Forbidden'
+            }
+
+        # TODO: This should be a method in the Job model
+        def _delete_job(app, job_id):
+            with app.app_context():
+                job = Job.query.get(job_id)
+                job.delete()
+                db.session.commit()
+
+        # Delete the job in a background task
+        socketio.start_background_task(
+            target=_delete_job,
+            app=current_app._get_current_object(),
+            job_id=job_id
+        )
+
+        return {
+            'code': 202,
+            'body': f'Job "{job.title}" marked for deletion'
+        }
+
+
+    @admin_required
+    def on_get_log(self, job_hashid: str):
+        # Decode the job hashid
+        job_id = hashids.decode(job_hashid)
+
+        # Validate job_id
+        if not isinstance(job_id, int):
+            return {
+                'code': 400,
+                'body': 'job_id is invalid'
+            }
+
+        # Load job from database
+        job = Job.query.get(job_id)
+        if job is None:
+            return {
+                'code': 404,
+                'body': 'Job not found'
+            }
+
+        # Check if the job is already processed
+        if job.status not in [JobStatus.COMPLETED, JobStatus.FAILED]:
+            return {
+                'code': 409,
+                'body': 'Job is not done processing'
+            }
+
+        # Read the log file
+        with open(job.path / 'pipeline_data' / 'logs' / 'pyflow_log.txt') as log_file:
+            job_log = log_file.read()
+
+        return {
+            'code': 200,
+            'body': job_log
+        }
+
+
+    @login_required
+    def on_restart(self, job_hashid: str):
+        # Decode the job hashid
+        job_id = hashids.decode(job_hashid)
+
+        # Validate job_id
+        if not isinstance(job_id, int):
+            return {
+                'code': 400,
+                'body': 'job_id is invalid'
+            }
+
+        # Load job from database
+        job = Job.query.get(job_id)
+        if job is None:
+            return {
+                'code': 404,
+                'body': 'Job not found'
+            }
+
+        # Check if the current user is allowed to restart the job
+        if not (job.user == current_user or current_user.is_administrator):
+            return {
+                'code': 403,
+                'body': 'Forbidden'
+            }
+
+        # TODO: This should be a method in the Job model
+        def _restart_job(app, job_id):
+            with app.app_context():
+                job = Job.query.get(job_id)
+                job.restart()
+                db.session.commit()
+
+        # Restart the job in a background task
+        socketio.start_background_task(
+            target=_restart_job,
+            app=current_app._get_current_object(),
+            job_id=job_id
+        )
+
+        return {
+            'code': 202,
+            'body': f'Job "{job.title}" restarted'
+        }

app/models/corpus.py

@@ -9,7 +9,7 @@ import shutil
 import xml.etree.ElementTree as ET
 from app import db
 from app.converters.vrt import normalize_vrt_file
-from app.extensions.sqlalchemy import IntEnumColumn
+from app.extensions.sqlalchemy_extras import IntEnumColumn
 from .corpus_follower_association import CorpusFollowerAssociation
 
 

app/models/job.py

@@ -7,7 +7,7 @@ from typing import Union
 from pathlib import Path
 import shutil
 from app import db
-from app.extensions.sqlalchemy import ContainerColumn, IntEnumColumn
+from app.extensions.sqlalchemy_extras import ContainerColumn, IntEnumColumn
 
 
 class JobStatus(IntEnum):

app/models/spacy_nlp_pipeline_model.py

@@ -5,7 +5,7 @@ from pathlib import Path
 import requests
 import yaml
 from app import db
-from app.extensions.sqlalchemy import ContainerColumn
+from app.extensions.sqlalchemy_extras import ContainerColumn
 from .file_mixin import FileMixin
 from .user import User
 

app/models/tesseract_ocr_pipeline_model.py

@@ -5,7 +5,7 @@ from pathlib import Path
 import requests
 import yaml
 from app import db
-from app.extensions.sqlalchemy import ContainerColumn
+from app.extensions.sqlalchemy_extras import ContainerColumn
 from .file_mixin import FileMixin
 from .user import User
 

app/models/user.py

@@ -12,7 +12,7 @@ import re
 import secrets
 import shutil
 from app import db, hashids
-from app.extensions.sqlalchemy import IntEnumColumn
+from app.extensions.sqlalchemy_extras import IntEnumColumn
 from .corpus import Corpus
 from .corpus_follower_association import CorpusFollowerAssociation
 from .corpus_follower_role import CorpusFollowerRole

app/users/events.py

@@ -1,12 +1,12 @@
 from flask_login import current_user
 from flask_socketio import join_room, leave_room
 from app import hashids, socketio
-from app.decorators import socketio_login_required
+from app.extensions.flask_socketio_extras import login_required
 from app.models import User
 
 
 @socketio.on('GET /users/<user_id>')
-@socketio_login_required
+@login_required
 def get_user(user_hashid):
     user_id = hashids.decode(user_hashid)
     user = User.query.get(user_id)
@@ -22,7 +22,7 @@ def get_user(user_hashid):
 
 
 @socketio.on('SUBSCRIBE /users/<user_id>')
-@socketio_login_required
+@login_required
 def subscribe_user(user_hashid):
     user_id = hashids.decode(user_hashid)
     user = User.query.get(user_id)
@@ -35,7 +35,7 @@ def subscribe_user(user_hashid):
 
 
 @socketio.on('UNSUBSCRIBE /users/<user_id>')
-@socketio_login_required
+@login_required
 def unsubscribe_user(user_hashid):
     user_id = hashids.decode(user_hashid)
     user = User.query.get(user_id)

app/users/nevents.py

@@ -1,12 +1,12 @@
 from flask_login import current_user
-from flask_socketio import join_room, leave_room
+from flask_socketio import join_room
 from app import hashids, socketio
-from app.decorators import socketio_admin_required, socketio_login_required
+from app.extensions.flask_socketio_extras import admin_required, login_required
 from app.models import User
 
 
 @socketio.on('GET /users')
-@socketio_admin_required
+@admin_required
 def get_users():
     users = User.query.filter_by().all()
     return {
@@ -20,14 +20,14 @@ def get_users():
 
 
 @socketio.on('SUBSCRIBE /users')
-@socketio_admin_required
+@admin_required
 def subscribe_users():
     join_room('/users')
     return {'options': {'status': 200, 'statusText': 'OK'}}
 
 
 @socketio.on('GET /users/<user_id>')
-@socketio_login_required
+@login_required
 def get_user(user_hashid):
     user_id = hashids.decode(user_hashid)
     user = User.query.get(user_id)
@@ -46,7 +46,7 @@ def get_user(user_hashid):
 
 
 @socketio.on('SUBSCRIBE /users/<user_id>')
-@socketio_login_required
+@login_required
 def subscribe_user(user_hashid):
     user_id = hashids.decode(user_hashid)
     user = User.query.get(user_id)
@@ -59,7 +59,7 @@ def subscribe_user(user_hashid):
 
 
 @socketio.on('GET /public_users')
-@socketio_login_required
+@login_required
 def get_public_users():
     users = User.query.filter_by(is_public=True).all()
     return {
@@ -76,14 +76,14 @@ def get_public_users():
 
 
 @socketio.on('SUBSCRIBE /users')
-@socketio_admin_required
+@admin_required
 def subscribe_users():
     join_room('/public_users')
     return {'options': {'status': 200, 'statusText': 'OK'}}
 
 
 @socketio.on('GET /public_users/<user_id>')
-@socketio_login_required
+@login_required
 def get_user(user_hashid):
     user_id = hashids.decode(user_hashid)
     user = User.query.filter_by(id=user_id, is_public=True).first()
@@ -102,7 +102,7 @@ def get_user(user_hashid):
 
 
 @socketio.on('SUBSCRIBE /public_users/<user_id>')
-@socketio_login_required
+@login_required
 def subscribe_user(user_hashid):
     user_id = hashids.decode(user_hashid)
     user = User.query.filter_by(id=user_id, is_public=True).first()

app/users/settings/forms.py

@@ -16,7 +16,7 @@ from wtforms.validators import (
     Regexp
 )
 from app.models import User, UserSettingJobStatusMailNotificationLevel
-from app.extensions.wtforms.validators import FileSize
+from app.extensions.wtforms_extras.validators import FileSize
 
 
 class UpdateAccountInformationForm(FlaskForm):