-
Notifications
You must be signed in to change notification settings - Fork 28
/
Copy pathserver.py
95 lines (71 loc) · 2.81 KB
/
server.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
from flask import render_template, request, redirect
from flask_security import Security, logout_user, login_required
from flask_security.utils import encrypt_password, verify_password
from flask_restless import APIManager
from flask_jwt import JWT, jwt_required
from database import db
from application import app
from models import User, SomeStuff, user_datastore
from admin import init_admin
# Setup Flask-Security =======================================================
security = Security(app, user_datastore)
# Views ======================================================================
@app.route('/')
def home():
return render_template('index.html')
@app.route('/mypage')
@login_required
def mypage():
return render_template('mypage.html')
@app.route('/logout')
def log_out():
logout_user()
return redirect(request.args.get('next') or '/')
# JWT Token authentication ===================================================
def authenticate(username, password):
user = user_datastore.find_user(email=username)
if user and username == user.email and verify_password(password, user.password):
return user
return None
def load_user(payload):
user = user_datastore.find_user(id=payload['identity'])
return user
jwt = JWT(app, authenticate, load_user)
# Flask-Restless API =========================================================
@jwt_required()
def auth_func(**kw):
pass
apimanager = APIManager(app, flask_sqlalchemy_db=db)
apimanager.create_api(SomeStuff,
methods=['GET', 'POST', 'DELETE', 'PUT'],
url_prefix='/api/v1',
collection_name='free_stuff',
include_columns=['id', 'data1', 'data2', 'user_id'])
apimanager.create_api(SomeStuff,
methods=['GET', 'POST', 'DELETE', 'PUT'],
url_prefix='/api/v1',
preprocessors=dict(GET_SINGLE=[auth_func], GET_MANY=[auth_func]),
collection_name='protected_stuff',
include_columns=['id', 'data1', 'data2', 'user_id'])
# Setup Admin ================================================================
init_admin()
# Bootstrap ==================================================================
def create_test_models():
user_datastore.create_user(email='test', password=encrypt_password('test'))
user_datastore.create_user(email='test2', password=encrypt_password('test2'))
stuff = SomeStuff(data1=2, data2='toto', user_id=1)
db.session.add(stuff)
stuff = SomeStuff(data1=5, data2='titi', user_id=1)
db.session.add(stuff)
db.session.commit()
@app.before_first_request
def bootstrap_app():
if not app.config['TESTING']:
if db.session.query(User).count() == 0:
create_test_models()
# Start server ===============================================================
if __name__ == '__main__':
db.init_app(app)
with app.app_context():
db.create_all()
app.run()