-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest-app.py
78 lines (62 loc) · 2.46 KB
/
test-app.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
from flask import Flask, render_template,request
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import create_engine
from datetime import datetime
from db import db
from Cipher import _encrypt, _decrypt, CRYPTO_KEY, CRYPTO_IV
from sqlalchemy.sql import text
import psycopg2
import urllib.parse as urlparse
import binascii
import os
db = SQLAlchemy()
class User(db.Model):
user_id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.Text, unique=True, nullable=False)
password = db.Column(db.Text, nullable=False)
spotify_id = db.Column(db.Text, unique=True, nullable=False)
date_created = db.Column(db.DateTime, nullable=False)
date_updated = db.Column(db.TIMESTAMP, nullable=False)
city = db.Column(db.Text, nullable=False)
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = os.environ['DB_URI']
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db.init_app(app)
@app.route('/', methods=['GET','POST'])
def login():
error = ''
engine = create_engine(os.environ['DB_URI'])
connection = engine.connect()
if request.method == 'POST':
email = request.form['email']
password = request.form['password']
cipher_text = _encrypt(password, CRYPTO_KEY, CRYPTO_IV)
select_query = """SELECT * FROM "User" WHERE email=:test1 AND password=:test2;"""
User_data = connection.execute(text(select_query), test1 = email, test2 = cipher_text.decode('utf-8') ).fetchone()
if not User_data and email != None:
error = 'Wrong email or password'
else:
spotify_id = User_data[3]
date_created = User_data[4]
date_updated = User_data[5]
city = User_data[6]
user = User(email=email,
password=password,
spotify_id=spotify_id,
date_created=date_created,
date_updated=date_updated,
city=city)
return 'Welcome Back %s' % (user.email)
return render_template('login.html', error=error)
@app.route('/Users')
def contacts():
engine = create_engine(os.environ['DB_URI'])
connection = engine.connect()
select_query = 'SELECT * FROM "User"'
rows = connection.execute( select_query ).fetchall()
my_list = []
for row in rows:
my_list.append(row[1])
return render_template('template.html', results = my_list)
if __name__ == '__main__':
app.run(threaded=true)