viveksantayana/ska-referee-test
1
0
Fork 0
Code Issues 1 Pull Requests Projects Releases 19 Wiki Activity

Exception handling for database queries

Browse Source
This commit is contained in:
Vivek Santayana
2022-08-20 10:56:43 +01:00
parent b8fd65d856
commit 866c9b10cf
15 changed files with 234 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

114
ref-test/app/admin/views.py
View File

@ -1,13 +1,15 @@
from ..forms.admin import AddTimeAdjustment, CreateTest, CreateUser, DeleteUser, Login, Register, ResetPassword, UpdatePassword, UpdateUser, UploadData
from ..models import Dataset, Entry, Test, User
from ..tools.auth import disable_if_logged_in, require_account_creation
from ..tools.forms import get_dataset_choices, get_time_options, send_errors_to_client
from ..tools.data import check_dataset_exists, check_is_json, validate_json
from ..tools.forms import get_dataset_choices, get_time_options, send_errors_to_client
from ..tools.logs import write
from ..tools.test import answer_options, get_correct_answers
from flask import abort, Blueprint, jsonify, render_template, redirect, request, send_file, session
from flask.helpers import flash, url_for
from flask import abort, Blueprint, jsonify, render_template, request, send_file, session
from flask.helpers import abort, flash, redirect, url_for
from flask_login import current_user, login_required
from sqlalchemy.exc import SQLAlchemyError
from datetime import date, datetime, timedelta
from json import loads
@ -26,8 +28,12 @@ admin = Blueprint(
@admin.route('/dashboard/')
@login_required
def _home():
tests = Test.query.all()
results = Entry.query.all()
try:
tests = Test.query.all()
results = Entry.query.all()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
current_tests = [ test for test in tests if test.end_date >= datetime.now() and test.start_date.date() <= date.today() ]
current_tests.sort(key= lambda x: x.end_date, reverse=True)
upcoming_tests = [ test for test in tests if test.start_date.date() > datetime.now().date()]
@ -39,8 +45,12 @@ def _home():
@admin.route('/settings/')
@login_required
def _settings():
users = User.query.all()
datasets = Dataset.query.all()
try:
users = User.query.all()
datasets = Dataset.query.all()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
return render_template('/admin/settings/index.html', users=users, datasets=datasets)
@admin.route('/login/', methods=['GET','POST'])
@ -50,7 +60,10 @@ def _login():
form = Login()
if request.method == 'POST':
if form.validate_on_submit():
users = User.query.all()
try: users = User.query.all()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
user = None
for _user in users:
if _user.get_username() == request.form.get('username').lower():
@ -99,7 +112,10 @@ def _reset():
if request.method == 'POST':
if form.validate_on_submit():
user = None
users = User.query.all()
try: users = User.query.all()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
for _user in users:
if _user.get_username() == request.form.get('username'):
user = _user
@ -111,7 +127,10 @@ def _reset():
token = request.args.get('token')
if token:
user = User.query.filter_by(reset_token=token).first()
try: user = User.query.filter_by(reset_token=token).first()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
if not user: return redirect(url_for('admin._reset'))
verification_token = user.verification_token
user.clear_reset_tokens()
@ -128,7 +147,10 @@ def _update_password():
form = UpdatePassword()
if form.validate_on_submit():
user = session.pop('user')
user = User.query.filter_by(id=user).first()
try: user = User.query.filter_by(id=user).first()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
user.update(password=request.form.get('password'))
session['remembered_username'] = user.get_username()
flash('Your password has been reset.', 'success')
@ -139,7 +161,10 @@ def _update_password():
@login_required
def _users():
form = CreateUser()
users = User.query.all()
try: users = User.query.all()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
if request.method == 'POST':
if form.validate_on_submit():
password = request.form.get('password')
@ -156,7 +181,10 @@ def _users():
@admin.route('/settings/users/delete/<string:id>', methods=['GET', 'POST'])
@login_required
def _delete_user(id:str):
user = User.query.filter_by(id=id).first()
try: user = User.query.filter_by(id=id).first()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
form = DeleteUser()
if request.method == 'POST':
if not user: return jsonify({'error': 'User does not exist.'}), 400
@ -180,7 +208,10 @@ def _delete_user(id:str):
@admin.route('/settings/users/update/<string:id>', methods=['GET', 'POST'])
@login_required
def _update_user(id:str):
user = User.query.filter_by(id=id).first()
try: user = User.query.filter_by(id=id).first()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
form = UpdateUser()
if request.method == 'POST':
if not user: return jsonify({'error': 'User does not exist.'}), 400
@ -222,7 +253,10 @@ def _questions():
return jsonify({'error': message}), 400
return send_errors_to_client(form=form)
data = Dataset.query.all()
try: data = Dataset.query.all()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
return render_template('/admin/settings/questions.html', form=form, data=data)
@admin.route('/settings/questions/delete/', methods=['POST'])
@ -231,7 +265,10 @@ def _edit_questions():
id = request.get_json()['id']
action = request.get_json()['action']
if not action == 'delete': return jsonify({'error': 'Invalid action.'}), 400
dataset = Dataset.query.filter_by(id=id).first()
try: dataset = Dataset.query.filter_by(id=id).first()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
if action == 'delete': success, message = dataset.delete()
if success: return jsonify({'success': message}), 200
return jsonify({'error': message}), 400
@ -239,7 +276,10 @@ def _edit_questions():
@admin.route('/settings/questions/download/<string:id>/')
@login_required
def _download(id:str):
dataset = Dataset.query.filter_by(id=id).first()
try: dataset = Dataset.query.filter_by(id=id).first()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
if not dataset: return abort(404)
data_path = path.abspath(dataset.get_file())
return send_file(data_path, as_attachment=True, attachment_filename=f'{dataset.get_name()}.json')
@ -250,7 +290,10 @@ def _download(id:str):
@check_dataset_exists
def _tests(filter:str=None):
tests = None
_tests = Test.query.all()
try: _tests = Test.query.all()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
form = None
now = datetime.now()
if filter not in ['create','active','scheduled','expired','all']: return redirect(url_for('admin._tests', filter='active'))
@ -296,7 +339,10 @@ def _create_test():
new_test.end_date = datetime.strptime(new_test.end_date, '%Y-%m-%dT%H:%M')
new_test.time_limit = None if request.form.get('time_limit') == 'none' else int(request.form.get('time_limit'))
dataset = request.form.get('dataset')
new_test.dataset = Dataset.query.filter_by(id=dataset).first()
try: new_test.dataset = Dataset.query.filter_by(id=dataset).first()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
success, message = new_test.create()
if success:
flash(message=message, category='success')
@ -310,7 +356,10 @@ def _edit_test():
id = request.get_json()['id']
action = request.get_json()['action']
if action not in ['start', 'delete', 'end']: return jsonify({'error': 'Invalid action.'}), 400
test = Test.query.filter_by(id=id).first()
try: test = Test.query.filter_by(id=id).first()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
if not test: return jsonify({'error': 'Could not find the corresponding test to delete.'}), 404
if action == 'delete': success, message = test.delete()
if action == 'start': success, message = test.start()
@ -324,7 +373,10 @@ def _edit_test():
@login_required
def _view_test(id:str=None):
form = AddTimeAdjustment()
test = Test.query.filter_by(id=id).first()
try: test = Test.query.filter_by(id=id).first()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
if request.method == 'POST':
if not test: return jsonify({'error': 'Invalid test ID.'}), 404
if form.validate_on_submit():
@ -341,7 +393,10 @@ def _view_test(id:str=None):
@admin.route('/test/<string:id>/delete-adjustment/', methods=['POST'])
@login_required
def _delete_adjustment(id:str=None):
test = Test.query.filter_by(id=id).first()
try: test = Test.query.filter_by(id=id).first()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
if not test: return jsonify({'error': 'Invalid test ID.'}), 404
user_code = request.get_json()['user_code'].lower()
success, message = test.remove_adjustment(user_code)
@ -351,13 +406,19 @@ def _delete_adjustment(id:str=None):
@admin.route('/results/')
@login_required
def _view_entries():
entries = Entry.query.all()
try: entries = Entry.query.all()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
return render_template('/admin/results.html', entries = entries)
@admin.route('/results/<string:id>/', methods = ['GET', 'POST'])
@login_required
def _view_entry(id:str=None):
entry = Entry.query.filter_by(id=id).first()
try: entry = Entry.query.filter_by(id=id).first()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
if request.method == 'POST':
if not entry: return jsonify({'error': 'Invalid entry ID.'}), 404
action = request.get_json()['action']
@ -388,6 +449,9 @@ def _view_entry(id:str=None):
def _generate_certificate():
from ..extensions import db
id = request.get_json()['id']
entry = Entry.query.filter_by(id=id).first()
try: entry = Entry.query.filter_by(id=id).first()
except SQLAlchemyError as exception:
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
return abort(500)
if not entry: return jsonify({'error': 'Invalid entry ID.'}), 404
return render_template('/admin/components/certificate.html', entry = entry)