214 lines
7.2 KiB
Python
214 lines
7.2 KiB
Python
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/<update_type>', 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()
|