-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathsql.py
65 lines (49 loc) · 2.08 KB
/
sql.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
from __init__ import db
from model import Users
import random
# this is method called by frontend, it has been randomized between Alchemy and Native SQL for fun
def users_all():
if random.randint(0, 1) == 0:
table = users_all_alc()
else:
table = users_all_sql()
return table
# SQLAlchemy extract all users from database
def users_all_alc():
table = Users.query.all()
json_ready = [peep.read() for peep in table]
return json_ready
# Native SQL extract all users from database
def users_all_sql():
table = db.session.execute('select * from users')
json_ready = sqlquery_2_list(table)
return json_ready
# SQLAlchemy extract users from database matching term
def users_ilike(term):
"""filter Users table by term into JSON list (ordered by User.name)"""
term = "%{}%".format(term) # "ilike" is case insensitive and requires wrapped %term%
table = Users.query.order_by(Users.name).filter((Users.name.ilike(term)) | (Users.email.ilike(term)))
return [peep.read() for peep in table]
# SQLAlchemy extract single user from database matching ID
def user_by_id(userid):
"""finds User in table matching userid """
return Users.query.filter_by(userID=userid).first()
# SQLAlchemy extract single user from database matching email
def user_by_email(email):
"""finds User in table matching email """
return Users.query.filter_by(email=email).first()
# ALGORITHM to convert the results of an SQL Query to a JSON ready format in Python
def sqlquery_2_list(rows):
out_list = []
keys = rows.keys() # "Keys" are the columns of the sql query
for values in rows: # "Values" are rows within the SQL database
row_dictionary = {}
for i in range(len(keys)): # This loop lines up K, V pairs, same as JSON style
row_dictionary[keys[i]] = values[i]
row_dictionary["query"] = "by_sql" # This is for fun a little watermark
out_list.append(row_dictionary) # Finally we have a out_list row
return out_list
# Test queries
if __name__ == "__main__":
for i in range(10):
print(users_all())