66 lines
1.9 KiB
Python
66 lines
1.9 KiB
Python
|
|
from ..extensions import db
|
|
from ..tools.logs import write
|
|
|
|
from flask import jsonify
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from wtforms.validators import ValidationError
|
|
|
|
import json
|
|
from sqlalchemy.ext import mutable
|
|
|
|
class JsonEncodedDict(db.TypeDecorator):
|
|
"""Enables JSON storage by encoding and decoding on the fly."""
|
|
impl = db.Text
|
|
|
|
def process_bind_param(self, value, dialect):
|
|
if value is None:
|
|
return '{}'
|
|
else:
|
|
return json.dumps(value)
|
|
|
|
def process_result_value(self, value, dialect):
|
|
if value is None:
|
|
return {}
|
|
else:
|
|
return json.loads(value)
|
|
|
|
mutable.MutableDict.associate_with(JsonEncodedDict)
|
|
|
|
def value(min:int=0, max:int=None):
|
|
if not max:
|
|
message = f'Value must be greater than {min}.'
|
|
else:
|
|
message = f'Value must be between {min} and {max}.'
|
|
def length(form, field):
|
|
value = field.data or 0
|
|
if value < min or max != None and value > max:
|
|
raise ValidationError(message)
|
|
return length
|
|
|
|
def get_time_options():
|
|
time_options = [
|
|
('none', 'None'),
|
|
('60', '1 hour'),
|
|
('90', '1 hour 30 minutes'),
|
|
('120', '2 hours')
|
|
]
|
|
return time_options
|
|
|
|
def get_dataset_choices():
|
|
from ..models import Dataset
|
|
try: datasets = Dataset.query.all()
|
|
except (SQLAlchemyError, ConnectionError) as exception:
|
|
write('system.log', f'Database error when fetching dataset lists: {exception}')
|
|
return []
|
|
dataset_choices = []
|
|
for dataset in datasets:
|
|
label = dataset.get_name()
|
|
label = f'{label} (Default)' if dataset.default else label
|
|
choice = (dataset.id, label)
|
|
dataset_choices.append(choice)
|
|
return dataset_choices
|
|
|
|
def send_errors_to_client(form):
|
|
errors = [*form.errors.values()]
|
|
return jsonify({ 'error': errors}), 400 |