nopaque/app/models.py
2020-05-11 16:09:09 +02:00

511 lines
17 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from datetime import datetime
from flask import current_app
from flask_login import UserMixin, AnonymousUserMixin
from itsdangerous import BadSignature, TimedJSONWebSignatureSerializer
from werkzeug.security import generate_password_hash, check_password_hash
from werkzeug.utils import secure_filename
from . import db, login_manager
class Permission:
"""
Defines User permissions as integers by the power of 2. User permission
can be evaluated using the bitwise operator &. 3 equals to CREATE_JOB and
DELETE_JOB and so on.
"""
MANAGE_CORPORA = 1
MANAGE_JOBS = 2
# PERMISSION_NAME = 4
# PERMISSION_NAME = 8
ADMIN = 16
class Role(db.Model):
"""
Model for the different roles Users can have. Is a one-to-many
relationship. A Role can be associated with many User rows.
"""
__tablename__ = 'roles'
# Primary key
id = db.Column(db.Integer, primary_key=True)
# Fields
default = db.Column(db.Boolean, default=False, index=True)
name = db.Column(db.String(64), unique=True)
permissions = db.Column(db.BigInteger)
# Relationships
users = db.relationship('User', backref='role', lazy='dynamic')
def to_dict(self):
return {'id': self.id,
'default': self.default,
'name': self.name,
'permissions': self.permissions}
def __init__(self, **kwargs):
super(Role, self).__init__(**kwargs)
if self.permissions is None:
self.permissions = 0
def __repr__(self):
"""
String representation of the Role. For human readability.
"""
return '<Role %r>' % self.name
def add_permission(self, perm):
"""
Add new permission to Role. Input is a Permission.
"""
if not self.has_permission(perm):
self.permissions += perm
def remove_permission(self, perm):
"""
Removes permission from a Role. Input a Permission.
"""
if self.has_permission(perm):
self.permissions -= perm
def reset_permissions(self):
"""
Resets permissions to zero. Zero equals no permissions at all.
"""
self.permissions = 0
def has_permission(self, perm):
"""
Checks if a Role has a specific Permission. Does this with the bitwise
operator.
"""
return self.permissions & perm == perm
@staticmethod
def insert_roles():
"""
Inserts roles into the database. This has to be executed befor Users
are added to the database. Otherwiese Users will not have a Role
assigned to them. Order of the roles dictionary determines the ID of
each role. Users have the ID 1 and Administrators have the ID 2.
"""
roles = {'User': [Permission.MANAGE_CORPORA, Permission.MANAGE_JOBS],
'Administrator': [Permission.MANAGE_CORPORA,
Permission.MANAGE_JOBS, Permission.ADMIN]}
default_role = 'User'
for r in roles:
role = Role.query.filter_by(name=r).first()
if role is None:
role = Role(name=r)
role.reset_permissions()
for perm in roles[r]:
role.add_permission(perm)
role.default = (role.name == default_role)
db.session.add(role)
db.session.commit()
class User(UserMixin, db.Model):
"""
Model for Users that are registered to Opaque.
"""
__tablename__ = 'users'
# Primary key
id = db.Column(db.Integer, primary_key=True)
# Foreign keys
role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))
# Fields
confirmed = db.Column(db.Boolean, default=False)
email = db.Column(db.String(254), unique=True, index=True)
last_seen = db.Column(db.DateTime(), default=datetime.utcnow)
member_since = db.Column(db.DateTime(), default=datetime.utcnow)
password_hash = db.Column(db.String(128))
setting_dark_mode = db.Column(db.Boolean, default=False)
setting_job_status_mail_notifications = db.Column(db.String(16),
default='end')
setting_job_status_site_notifications = db.Column(db.String(16),
default='all')
username = db.Column(db.String(64), unique=True, index=True)
# Relationships
corpora = db.relationship('Corpus', backref='creator', lazy='dynamic',
cascade='save-update, merge, delete')
jobs = db.relationship('Job', backref='creator', lazy='dynamic',
cascade='save-update, merge, delete')
def to_dict(self):
return {'id': self.id,
'role_id': self.role_id,
'confirmed': self.confirmed,
'email': self.email,
'last_seen': self.last_seen.timestamp(),
'member_since': self.member_since.timestamp(),
'username': self.username,
'settings': {'dark_mode': self.setting_dark_mode,
'job_status_mail_notifications':
self.setting_job_status_mail_notifications,
'job_status_site_notifications':
self.setting_job_status_site_notifications},
'corpora': {corpus.id: corpus.to_dict()
for corpus in self.corpora},
'jobs': {job.id: job.to_dict() for job in self.jobs}}
def __repr__(self):
"""
String representation of the User. For human readability.
"""
return '<User %r>' % self.username
def __init__(self, **kwargs):
super(User, self).__init__(**kwargs)
if self.role is None:
if self.email == current_app.config['NOPAQUE_ADMIN']:
self.role = Role.query.filter_by(name='Administrator').first()
if self.role is None:
self.role = Role.query.filter_by(default=True).first()
def generate_confirmation_token(self, expiration=3600):
"""
Generates a confirmation token for user confirmation via email.
"""
s = TimedJSONWebSignatureSerializer(current_app.config['SECRET_KEY'],
expiration)
return s.dumps({'confirm': self.id}).decode('utf-8')
def generate_reset_token(self, expiration=3600):
"""
Generates a reset token for password reset via email.
"""
s = TimedJSONWebSignatureSerializer(current_app.config['SECRET_KEY'],
expiration)
return s.dumps({'reset': self.id}).decode('utf-8')
def confirm(self, token):
"""
Confirms User if the given token is valid and not expired.
"""
s = TimedJSONWebSignatureSerializer(current_app.config['SECRET_KEY'])
try:
data = s.loads(token.encode('utf-8'))
except BadSignature:
return False
if data.get('confirm') != self.id:
return False
self.confirmed = True
db.session.add(self)
return True
@staticmethod
def reset_password(token, new_password):
"""
Resets password for User if the given token is valid and not expired.
"""
s = TimedJSONWebSignatureSerializer(current_app.config['SECRET_KEY'])
try:
data = s.loads(token.encode('utf-8'))
except BadSignature:
return False
user = User.query.get(data.get('reset'))
if user is None:
return False
user.password = new_password
db.session.add(user)
return True
@property
def password(self):
raise AttributeError('password is not a readable attribute')
@password.setter
def password(self, password):
self.password_hash = generate_password_hash(password)
def verify_password(self, password):
return check_password_hash(self.password_hash, password)
def can(self, perm):
"""
Checks if a User with its current role can doe something. Checks if the
associated role actually has the needed Permission.
"""
return self.role is not None and self.role.has_permission(perm)
def is_administrator(self):
"""
Checks if User has Admin permissions.
"""
return self.can(Permission.ADMIN)
def ping(self):
self.last_seen = datetime.utcnow()
db.session.add(self)
def delete(self):
"""
Delete the user and its corpora and jobs from database and filesystem.
"""
for job in self.jobs:
job.delete()
for corpus in self.corpora:
corpus.delete()
db.session.delete(self)
db.session.commit()
class AnonymousUser(AnonymousUserMixin):
"""
Model replaces the default AnonymousUser.
"""
def can(self, permissions):
return False
def is_administrator(self):
return False
class JobInput(db.Model):
"""
Class to define JobInputs.
"""
__tablename__ = 'job_inputs'
# Primary key
id = db.Column(db.Integer, primary_key=True)
# Foreign keys
job_id = db.Column(db.Integer, db.ForeignKey('jobs.id'))
# Fields
dir = db.Column(db.String(255))
filename = db.Column(db.String(255))
def __repr__(self):
"""
String representation of the JobInput. For human readability.
"""
return '<JobInput %r>' % self.filename
def to_dict(self):
return {'id': self.id,
'job_id': self.job_id,
'filename': self.filename}
class JobResult(db.Model):
"""
Class to define JobResults.
"""
__tablename__ = 'job_results'
# Primary key
id = db.Column(db.Integer, primary_key=True)
# Foreign keys
job_id = db.Column(db.Integer, db.ForeignKey('jobs.id'))
# Fields
dir = db.Column(db.String(255))
filename = db.Column(db.String(255))
def __repr__(self):
"""
String representation of the JobResult. For human readability.
"""
return '<JobResult %r>' % self.filename
def to_dict(self):
return {'id': self.id,
'job_id': self.job_id,
'filename': self.filename}
class Job(db.Model):
"""
Class to define Jobs.
"""
__tablename__ = 'jobs'
# Primary key
id = db.Column(db.Integer, primary_key=True)
# Foreign keys
user_id = db.Column(db.Integer, db.ForeignKey('users.id'))
# Fields
creation_date = db.Column(db.DateTime(), default=datetime.utcnow)
description = db.Column(db.String(255))
end_date = db.Column(db.DateTime())
mem_mb = db.Column(db.Integer)
n_cores = db.Column(db.Integer)
secure_filename = db.Column(db.String(32))
service = db.Column(db.String(64))
'''
' Service specific arguments as string list.
' Example: ["-l eng", "--keep-intermediates", "--skip-binarization"]
'''
service_args = db.Column(db.String(255))
service_version = db.Column(db.String(16))
status = db.Column(db.String(16))
title = db.Column(db.String(32))
# Relationships
inputs = db.relationship('JobInput', backref='job', lazy='dynamic',
cascade='save-update, merge, delete')
results = db.relationship('JobResult', backref='job', lazy='dynamic',
cascade='save-update, merge, delete')
notifications_data = db.relationship('NotificationData',
backref='job',
lazy='dynamic',
cascade='save-update, merge, delete')
def __repr__(self):
"""
String representation of the Job. For human readability.
"""
return '<Job %r>' % self.title
def create_secure_filename(self):
"""
Takes the job.title string nad cratesa a secure filename from this.
"""
self.secure_filename = secure_filename(self.title)
def delete(self):
"""
Delete the job and its inputs and results from the database.
"""
for input in self.inputs:
db.session.delete(input)
for result in self.results:
db.session.delete(result)
db.session.delete(self)
db.session.commit()
def to_dict(self):
return {'id': self.id,
'user_id': self.user_id,
'creation_date': self.creation_date.timestamp(),
'description': self.description,
'end_date': (self.end_date.timestamp() if self.end_date else
None),
'inputs': {input.id: input.to_dict() for input in self.inputs},
'mem_mb': self.mem_mb,
'n_cores': self.n_cores,
'results': {result.id: result.to_dict()
for result in self.results},
'service': self.service,
'service_args': self.service_args,
'service_version': self.service_version,
'status': self.status,
'title': self.title}
class NotificationData(db.Model):
"""
Class to define notification data used for sending a notification mail with
nopaque_notify.
"""
__tablename__ = 'notifications_data'
# Primary key
id = db.Column(db.Integer, primary_key=True)
# Foreign Key
job_id = db.Column(db.Integer, db.ForeignKey('jobs.id'))
# Fields
notified_on_submitted = db.Column(db.Boolean, default=False)
notified_on_queued = db.Column(db.Boolean, default=False)
notified_on_running = db.Column(db.Boolean, default=False)
notified_on_complete = db.Column(db.Boolean, default=False)
notified_on_canceling = db.Column(db.Boolean, default=False)
class CorpusFile(db.Model):
"""
Class to define Files.
"""
__tablename__ = 'corpus_files'
# Primary key
id = db.Column(db.Integer, primary_key=True)
# Foreign keys
corpus_id = db.Column(db.Integer, db.ForeignKey('corpora.id'))
# Fields
address = db.Column(db.String(255))
author = db.Column(db.String(255))
booktitle = db.Column(db.String(255))
chapter = db.Column(db.String(255))
dir = db.Column(db.String(255))
editor = db.Column(db.String(255))
filename = db.Column(db.String(255))
institution = db.Column(db.String(255))
journal = db.Column(db.String(255))
pages = db.Column(db.String(255))
publisher = db.Column(db.String(255))
publishing_year = db.Column(db.Integer)
school = db.Column(db.String(255))
title = db.Column(db.String(255))
def delete(self):
self.corpus.status = 'unprepared'
db.session.delete(self)
db.session.commit()
def to_dict(self):
return {'id': self.id,
'corpus_id': self.corpus_id,
'address': self.address,
'author': self.author,
'booktitle': self.booktitle,
'chapter': self.chapter,
'editor': self.editor,
'filename': self.filename,
'institution': self.institution,
'journal': self.journal,
'pages': self.pages,
'publisher': self.publisher,
'publishing_year': self.publishing_year,
'school': self.school,
'title': self.title}
class Corpus(db.Model):
"""
Class to define a corpus.
"""
__tablename__ = 'corpora'
# Primary key
id = db.Column(db.Integer, primary_key=True)
# Foreign keys
user_id = db.Column(db.Integer, db.ForeignKey('users.id'))
# Fields
creation_date = db.Column(db.DateTime(), default=datetime.utcnow)
current_nr_of_tokens = db.Column(db.BigInteger, default=0)
description = db.Column(db.String(255))
last_edited_date = db.Column(db.DateTime(), default=datetime.utcnow)
max_nr_of_tokens = db.Column(db.BigInteger, default=2147483647)
status = db.Column(db.String(16))
title = db.Column(db.String(32))
# Relationships
files = db.relationship('CorpusFile', backref='corpus', lazy='dynamic',
cascade='save-update, merge, delete')
def to_dict(self):
return {'id': self.id,
'user_id': self.user_id,
'creation_date': self.creation_date.timestamp(),
'description': self.description,
'status': self.status,
'last_edited_date': self.last_edited_date.timestamp(),
'title': self.title,
'files': {file.id: file.to_dict() for file in self.files}}
def delete(self):
for corpus_file in self.files:
db.session.delete(corpus_file)
db.session.delete(self)
db.session.commit()
def __repr__(self):
"""
String representation of the corpus. For human readability.
"""
return '<Corpus %r>' % self.title
'''
' Flask-Login is told to use the applications custom anonymous user by setting
' its class in the login_manager.anonymous_user attribute.
'''
login_manager.anonymous_user = AnonymousUser
@login_manager.user_loader
def load_user(user_id):
return User.query.get(int(user_id))