-
Notifications
You must be signed in to change notification settings - Fork 0
/
app.py
268 lines (217 loc) · 8.17 KB
/
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
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
#!venv/bin/python
import os
import flask_admin
from flask import Flask, abort, redirect, render_template, request, url_for
from flask_admin import BaseView, expose
from flask_admin import helpers as admin_helpers
from flask_admin.contrib import sqla
from flask_security import (RoleMixin, Security, SQLAlchemyUserDatastore,
UserMixin, current_user, login_required)
from flask_security.utils import encrypt_password
from flask_sqlalchemy import SQLAlchemy
from werkzeug.utils import secure_filename
from app import app
import datetime
import json
import requests
# Create Flask application and initializing the DB
app = Flask(__name__)
app.config.from_pyfile('config.py')
db = SQLAlchemy(app)
# Define models
roles_users = db.Table(
'roles_users',
db.Column('user_id', db.Integer(), db.ForeignKey('user.id')),
db.Column('role_id', db.Integer(), db.ForeignKey('role.id'))
)
class Role(db.Model, RoleMixin):
id = db.Column(db.Integer(), primary_key=True)
name = db.Column(db.String(80), unique=True, nullable=False)
description = db.Column(db.String(255))
def __str__(self):
return self.name
class User(db.Model, UserMixin):
id = db.Column(db.Integer, primary_key=True)
first_name = db.Column(db.String(255))
last_name = db.Column(db.String(255))
email = db.Column(db.String(255), unique=True)
password = db.Column(db.String(255))
active = db.Column(db.Boolean())
confirmed_at = db.Column(db.DateTime())
roles = db.relationship('Role', secondary=roles_users,
backref=db.backref('users', lazy='dynamic'))
def __str__(self):
return self.email
# Setup Flask-Security
user_datastore = SQLAlchemyUserDatastore(db, User, Role)
security = Security(app, user_datastore)
# Create customized model view class
class MyModelView(sqla.ModelView):
def is_accessible(self):
if not current_user.is_active or not current_user.is_authenticated:
return False
if current_user.has_role('superuser'):
return True
return False
def _handle_view(self, name, **kwargs):
"""
Override builtin _handle_view in order to redirect users when a view is not accessible.
"""
if not self.is_accessible():
if current_user.is_authenticated:
# permission denied
abort(403)
else:
# login
return redirect(url_for('security.login', next=request.url))
# can_edit = True
edit_modal = True
create_modal = True
can_export = True
can_view_details = True
details_modal = True
class UserView(MyModelView):
column_editable_list = ['email', 'first_name', 'last_name']
column_searchable_list = column_editable_list
column_exclude_list = ['password']
# form_excluded_columns = column_exclude_list
column_details_exclude_list = column_exclude_list
column_filters = column_editable_list
# The node with which our application interacts, there can be multiple
# such nodes as well.
CONNECTED_NODE_ADDRESS = "http://127.0.0.1:8000"
posts = []
def fetch_posts():
"""
Function to fetch the chain from a blockchain node, parse the
data and store it locally.
"""
get_chain_address = "{}/chain".format(CONNECTED_NODE_ADDRESS)
response = requests.get(get_chain_address)
if response.status_code == 200:
content = []
chain = json.loads(response.content)
for block in chain["chain"]:
for tx in block["transactions"]:
tx["index"] = block["index"]
tx["hash"] = block["previous_hash"]
content.append(tx)
global posts
posts = sorted(content, key=lambda k: k['timestamp'],
reverse=True)
class CustomView(BaseView):
@expose('/')
def index(self):
fetch_posts()
return self.render('admin/custom_index.html',
posts=posts,
node_address=CONNECTED_NODE_ADDRESS,
readable_time=timestamp_to_string)
# Flask views
@app.route('/')
def index():
return render_template('index.html')
@app.route('/submit', methods=['POST'])
def submit_textarea():
"""
Endpoint to create a new transaction via our application.
"""
post_content = request.form["content"]
patient = request.form["patient"]
patient_id = request.form["patient_id"]
gender = request.form["gender"]
address = request.form["address"]
phone = request.form["phone"]
birthdate = request.form["birthdate"]
diagnose = request.form["diagnose"]
medications = request.form["medications"]
post_object = {
'patient': patient,
'patient_id': patient_id,
'gender': gender,
'address': address,
'phone': phone,
'birthdate': birthdate,
'diagnose': diagnose,
'medications': medications,
'content': post_content,
}
# Submit a transaction
new_tx_address = "{}/new_transaction".format(CONNECTED_NODE_ADDRESS)
requests.post(new_tx_address,
json=post_object,
headers={'Content-type': 'application/json'})
return redirect('/admin/custom/')
def timestamp_to_string(epoch_time):
return datetime.datetime.fromtimestamp(epoch_time).strftime('%H:%M')
# Create admin
admin = flask_admin.Admin(
app,
'My Dashboard',
base_template='my_master.html',
template_mode='bootstrap3',
)
# Add model views
admin.add_view(MyModelView(Role, db.session, menu_icon_type='fa', menu_icon_value='fa-server', name="Roles"))
admin.add_view(UserView(User, db.session, menu_icon_type='fa', menu_icon_value='fa-users', name="Users"))
admin.add_view(CustomView(name="Manage Patients", endpoint='custom', menu_icon_type='fa', menu_icon_value='fa-connectdevelop',))
# define a context processor for merging flask-admin's template context into the
# flask-security views.
@security.context_processor
def security_context_processor():
return dict(
admin_base_template=admin.base_template,
admin_view=admin.index_view,
h=admin_helpers,
get_url=url_for
)
def build_sample_db():
"""
Populate a small db with some example entries.
"""
import string
import random
db.drop_all()
db.create_all()
with app.app_context():
user_role = Role(name='user')
super_user_role = Role(name='superuser')
db.session.add(user_role)
db.session.add(super_user_role)
db.session.commit()
test_user = user_datastore.create_user(
first_name='Admin',
email='admin',
password=encrypt_password('admin'),
roles=[user_role, super_user_role]
)
first_names = [
'Harry', 'Amelia', 'Oliver', 'Jack', 'Isabella', 'Charlie', 'Sophie', 'Mia',
'Jacob', 'Thomas', 'Emily', 'Lily', 'Ava', 'Isla', 'Alfie', 'Olivia', 'Jessica',
'Riley', 'William', 'James', 'Geoffrey', 'Lisa', 'Benjamin', 'Stacey', 'Lucy'
]
last_names = [
'Brown', 'Smith', 'Patel', 'Jones', 'Williams', 'Johnson', 'Taylor', 'Thomas',
'Roberts', 'Khan', 'Lewis', 'Jackson', 'Clarke', 'James', 'Phillips', 'Wilson',
'Ali', 'Mason', 'Mitchell', 'Rose', 'Davis', 'Davies', 'Rodriguez', 'Cox', 'Alexander'
]
for i in range(len(first_names)):
tmp_email = first_names[i].lower() + "." + last_names[i].lower() + "@example.com"
tmp_pass = ''.join(random.choice(string.ascii_lowercase + string.digits) for i in range(10))
user_datastore.create_user(
first_name=first_names[i],
last_name=last_names[i],
email=tmp_email,
password=encrypt_password(tmp_pass),
roles=[user_role, ]
)
db.session.commit()
return
if __name__ == '__main__':
# Build a sample db on the fly, if one does not exist yet.
app_dir = os.path.realpath(os.path.dirname(__file__))
database_path = os.path.join(app_dir, app.config['DATABASE_FILE'])
if not os.path.exists(database_path):
build_sample_db()
# Start app
app.run(debug=True)