@ -2,9 +2,8 @@ import random
import operator
import json
import datetime
from functools import wraps
from flask import Flask , Response , redirect , url_for , request , session , abort , render_template
from flask_login import LoginManager
from flask import Flask , Response , redirect , url_for , request , abort , render_template
from flask_login import LoginManager , UserMixin , login_required , login_user , logout_user , current_user
import sqlalchemy
from sqlalchemy . sql import select , text
from dateutil . parser import parse
@ -20,6 +19,18 @@ app.config.update(
db_engine = { }
db_metadata = { }
meals = { }
users = { }
class User ( UserMixin ) :
def __init__ ( self , name ) :
self . name = name
self . id = - 1
@staticmethod
def get ( name ) :
if name not in users :
users [ name ] = User ( name )
return users [ name ]
def connect ( user , password , db , host = ' localhost ' , port = 5432 ) :
url = ' postgresql:// {} : {} @ {} : {} / {} '
@ -28,36 +39,6 @@ def connect(user, password, db, host='localhost', port=5432):
meta = sqlalchemy . MetaData ( bind = con , reflect = True )
return con , meta
def check_auth ( username , password ) :
query = " select id, password from users where username=:usrnm; "
r = db_engine . execute ( text ( query ) , usrnm = username )
results = [ ]
for row in r :
results . append ( dict ( row ) )
r . close ( )
if len ( results ) != 1 :
return False ;
#TODO: REMOVE TEMPORARY PASSWORD OVERRIDE VECTOR 1==1.
if results [ 0 ] [ ' password ' ] == password or 1 == 1 :
session [ ' uid ' ] = results [ 0 ] [ ' id ' ]
print ( session [ ' uid ' ] )
return True
else :
return False
def authenticate ( ) :
return Response ( ' Could not verify access level. Please retry ' , 401 , { ' WWW-Authenticate ' : ' Basic realm= " Login Required " ' } )
def requires_auth ( f ) :
@wraps ( f )
def decorated ( * args , * * kwargs ) :
auth = request . authorization
if not auth or not check_auth ( auth . username , auth . password ) :
return authenticate ( )
return f ( * args , * * kwargs )
return decorated
def getLeaderboard ( ) :
s = ' select u.username as name, u.score as score, max(m.meal_date) as last_meal from users as u left join meals as m on m.kuk = u.id group by u.id order by score, last_meal; '
r = db_engine . execute ( text ( s ) )
@ -97,29 +78,75 @@ def getUpcomingMeals():
# adding to meal
@app.route ( " /addme " )
@ requires_auth
@ login_required
def addme ( ) :
meal_id = request . args . get ( ' meal ' )
query = ' select :user_id = any (eaters) from meals where id =:meal_id; '
r = db_engine . execute ( text ( query ) , user_id = session[ ' uid ' ] , meal_id = meal_id )
r = db_engine . execute ( text ( query ) , user_id = current_user. id , meal_id = meal_id )
results = [ ]
for row in r :
results . append ( dict ( row ) )
if 1 == 1 :
return render_template ( ' appresponse.html ' , message = ' Have a nice meal ' + meal_id + " mister " + str ( session[ ' uid ' ] ) )
return render_template ( ' appresponse.html ' , message = ' Have a nice meal ' + meal_id + " mister " + str ( current_user. name ) )
else :
return render_template ( ' appresponse.html ' , message = ' Only one meal per person! ' )
@app.route ( ' / ' )
def index ( ) :
print ( getUpcomingMeals ( ) )
return render_template ( ' meal_list.html ' , leaderboard = getLeaderboard ( ) , meals = getUpcomingMeals ( ) , current_user_name = ' Wille ' )
@app.route ( ' /login ' , methods = [ ' GET ' , ' POST ' ] )
def login ( ) :
if request . method == ' POST ' :
username = request . form [ ' username ' ]
password = request . form [ ' password ' ]
r = db_engine . execute ( text ( ' select id from users where username=:username and password=:password ' ) , username = username , password = password )
result = [ ]
for row in r :
result . append ( dict ( row ) )
r . close ( )
print ( len ( users ) )
if len ( result ) > 0 :
u = User . get ( username )
u . id = result [ 0 ] [ ' id ' ]
login_user ( u )
return redirect ( request . args . get ( " next " ) )
return abort ( 401 )
else :
return Response ( '''
< form action = " " method = " post " >
< p > < input type = text name = username >
< p > < input type = password name = password >
< p > < input type = submit value = Login >
< / form >
''' )
with open ( ' admin.json ' ) as f :
postgres_credentials = json . load ( f )
db_engine , db_metadata = connect ( postgres_credentials [ ' username ' ] , postgres_credentials [ ' password ' ] , ' kuk_app ' )
meals = sqlalchemy . Table ( ' meals ' , db_metadata )
login_manager = LoginManager ( )
login_manager . init_app ( app )
login_manager . login_view = " login "
# somewhere to logout
@app.route ( " /logout " )
@login_required
def logout ( ) :
logout_user ( )
return Response ( ' <p>Logged out</p> ' )
# handle login failed
@app.errorhandler ( 401 )
def page_not_found ( e ) :
return Response ( ' <p>Login failed</p> ' )
# callback to reload the user object
@login_manager.user_loader
def load_user ( userid ) :
return User . get ( userid )
if __name__ == ' __main__ ' :
app . run ( ' 0.0.0.0 ' )
