from flask import Flask, Blueprint, request from flask_security import (SQLAlchemyUserDatastore, login_required, current_user, Security) from flask_security.utils import (verify_and_update_password, login_user, logout_user) from flask_restless import APIManager, ProcessingException from models import (db, User, Role, Gradeclass, Student, AttendanceUpdate, Presence, presense_map) from sqlalchemy import func import json from datetime import datetime, date from dateutil import parser app = Flask(__name__) app.config['DEBUG'] = True app.config['PORT'] = 5001 app.config['SECRET_KEY'] = '3kljk1232kl1j32kmk13nbciw' app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/db.sqlite' app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False app.config['SECURITY_PASSWORD_SALT'] = 'uaisfyasiduyaisiuf' app.config['SECURITY_UNAUTHORIZED_VIEW'] = None db.init_app(app) user_datastore = SQLAlchemyUserDatastore(db, User, Role) security = Security(app, user_datastore) def auth_func(*args, **kwargs): if not current_user.is_authenticated: raise ProcessingException(description='Not authenticated', code=401) return True verify_logged_id = dict(GET_SINGLE=[auth_func], GET_MANY=[auth_func]) manager = APIManager(app, flask_sqlalchemy_db=db) manager.create_api( Student, methods=['GET', 'PUT'], results_per_page=5, preprocessors=verify_logged_id) manager.create_api( Gradeclass, methods=['GET', 'PUT'], results_per_page=5, preprocessors=verify_logged_id) json_cont = {'Content-Type': 'application/json'} resp = lambda val, code: (json.dumps({'status': val}), code, json_cont) resp_res = lambda val, code: (json.dumps({'results': val}), code, json_cont) @app.before_first_request def create_test_data(): try: db.drop_all() db.create_all() for i in range(3): g_cls = Gradeclass("Class {}".format(i)) db.session.add(g_cls) db.session.commit() for s in range(5): stu = Student("TestStudent#{}-Class{}".format(s + 1, g_cls.id), g_cls.id) db.session.add(stu) user_datastore.create_user( email='user@example.com', password='password') db.session.commit() except: pass def parse_time(time_str): utc_now = (time_str and time_str == 'now') or not time_str time = datetime.utcnow() if utc_now else None try: time = parser.parse(time_str) except: pass return time def compute_attendance(request_type, object_id, start_time, end_time): def get_records(req_type, obj_id): return AttendanceUpdate.query.filter( AttendanceUpdate.update_type == req_type).filter( AttendanceUpdate.value_identifier == obj_id).filter( func.date(AttendanceUpdate.time) >= start_time.date()).filter( func.date(AttendanceUpdate.time) <= end_time.date()).all() def student_records(stud_id): stud_recs = get_records('student', stud_id) stud = Student.query.get(stud_id) gcls_recs = get_records('class', stud.gradeclass_id) pres_rec = stud_recs for gr in gcls_recs: # if any grade record overlaps with student record -> ignore if not any( map(lambda x: gr.time.date() == x.time.date(), stud_recs)): pres_rec.append(gr) recs = [] # sort student records by time for r in sorted(pres_rec, key=lambda x: x.time): recs.append({ "presence": presense_map[r.presence].lower(), "student": stud.student_name, "time": r.time.isoformat() }) return recs result = [] if request_type == 'student': result.extend(student_records(object_id)) if request_type == 'class': gcls = Gradeclass.query.get(object_id) # get result for each student for stud in gcls.students: result.extend(student_records(stud.id)) return result def upsert_attendance(update_type, object_id, attend_time, pres_enum): existing = AttendanceUpdate.query.filter( AttendanceUpdate.update_type == update_type).filter( AttendanceUpdate.value_identifier == object_id).filter( func.date( AttendanceUpdate.time) == attend_time.date()).first() if existing: existing.presence = pres_enum else: new_entry = AttendanceUpdate(update_type, object_id, attend_time, pres_enum) db.session.add(new_entry) db.session.commit() def validate_attend(update_type, object_id): if update_type in ['class', 'student'] and object_id: g_cls_found = update_type == 'class' and Gradeclass.query.get( object_id) stud_found = update_type == 'student' and Student.query.get(object_id) return g_cls_found or stud_found else: return False def get_attendance(update_type, request): object_id = request.args.get('identifier', None) start_timestr = request.args.get('start_time', None) end_timestr = request.args.get('end_time', None) start_time, end_time = parse_time(start_timestr), parse_time(end_timestr) ret_data = resp('invalid', 400) if validate_attend(update_type, object_id) and start_time and end_time: pres = compute_attendance(update_type, object_id, start_time, end_time) ret_data = resp_res(pres, 200) return ret_data def update_attendance(update_type, request): object_id = request.form.get('identifier', None) presence = request.form.get('presence', None) attend_str = request.form.get('time', None) attend_time = parse_time(attend_str) ret_data = resp('invalid', 400) if validate_attend( update_type, object_id) and presence and attend_time and presence.upper( ) in presense_map.values(): pres_enum = Presence[presence.upper()] upsert_attendance(update_type, object_id, attend_time, pres_enum) ret_data = resp('updated', 200) return ret_data @app.route('/api/attendance/', methods=['GET', 'POST']) @login_required def attendance(update_type): if request.method == 'GET': return get_attendance(update_type, request) elif request.method == 'POST': return update_attendance(update_type, request) @app.route('/api/login', methods=['POST']) def login(): username = request.form.get('username', None) password = request.form.get('password', None) ret_data = (json.dumps({'status': 'failed'}), 401, json_cont) if username and password: user = User.query.filter(User.email == username).first() if not user: ret_data = resp('notfound', 401) elif verify_and_update_password(password, user): login_user(user) ret_data = resp('success', 200) return ret_data @app.route('/api/logout', methods=['GET']) def logout(): logout_user() return resp('success', 200) @app.route('/') def home(): return 'welcome' if __name__ == '__main__': app.run()