mjgDi0Rx

1import os
2from flask import Flask, request, jsonify, make_response
3from flask_sqlalchemy import SQLAlchemy
4import uuid
5from werkzeug.security import generate_password_hash, check_password_hash
6import jwt
7import datetime
8from functools import wraps
9
10basedir = os.path.abspath(os.path.dirname(__file__))
11
12class Config(object):
13    SECRET_KEY = os.environ.get('SECRET_KEY') or 'supersecretkey'
14
15    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///' + os.path.join(basedir, 'app.db')
16
17    SQLALCHEMY_TRACK_MODIFICATIONS = False
18
19app = Flask(__name__)
20app.config.from_object(Config)
21db = SQLAlchemy(app)
22
23class User(db.Model):
24    id = db.Column(db.Integer, primary_key=True)
25    public_id = db.Column(db.String(50), unique=True)
26    name = db.Column(db.String(50))
27    password = db.Column(db.String(80))
28    admin = db.Column(db.Boolean)
29
30class Todo(db.Model):
31    id = db.Column(db.Integer, primary_key=True)
32    text = db.Column(db.String(50))
33    complete = db.Column(db.Boolean)
34    user_id = db.Column(db.Integer)
35
36def token_required(f):
37    @wraps(f)
38    def decorated(*args, **kwargs):
39        token = None
40
41        if 'x-access-token' in request.headers:
42            token = request.headers['x-access-token']
43
44        if not token:
45            return jsonify({'message' : 'Token is missing!'}), 401
46
47        try: 
48            data = jwt.decode(token, app.config['SECRET_KEY'])
49            current_user = User.query.filter_by(public_id=data['public_id']).first()
50        except:
51            return jsonify({'message' : 'Token is invalid!'}), 401
52
53        return f(current_user, *args, **kwargs)
54
55    return decorated
56
57@app.route('/user', methods=['GET'])
58@token_required
59def get_all_users(current_user):
60
61    if not current_user.admin:
62        return jsonify({'message' : 'Cannot perform that function!'})
63
64    users = User.query.all()
65
66    output = []
67
68    for user in users:
69        user_data = {}
70        user_data['public_id'] = user.public_id
71        user_data['name'] = user.name
72        user_data['password'] = user.password
73        user_data['admin'] = user.admin
74        output.append(user_data)
75
76    return jsonify({'users' : output})
77
78@app.route('/user/<public_id>', methods=['GET'])
79@token_required
80def get_one_user(current_user, public_id):
81
82    if not current_user.admin:
83        return jsonify({'message' : 'Cannot perform that function!'})
84
85    user = User.query.filter_by(public_id=public_id).first()
86
87    if not user:
88        return jsonify({'message' : 'No user found!'})
89
90    user_data = {}
91    user_data['public_id'] = user.public_id
92    user_data['name'] = user.name
93    user_data['password'] = user.password
94    user_data['admin'] = user.admin
95
96    return jsonify({'user' : user_data})
97
98@app.route('/user', methods=['POST'])
99@token_required
100def create_user(current_user):
101    if not current_user.admin:
102        return jsonify({'message' : 'Cannot perform that function!'})
103
104    data = request.get_json()
105
106    hashed_password = generate_password_hash(data['password'], method='sha256')
107
108    new_user = User(public_id=str(uuid.uuid4()), name=data['name'], password=hashed_password, admin=False)
109    db.session.add(new_user)
110    db.session.commit()
111
112    return jsonify({'message' : 'New user created!'})
113
114@app.route('/user/<public_id>', methods=['PUT'])
115@token_required
116def promote_user(current_user, public_id):
117    if not current_user.admin:
118        return jsonify({'message' : 'Cannot perform that function!'})
119
120    user = User.query.filter_by(public_id=public_id).first()
121
122    if not user:
123        return jsonify({'message' : 'No user found!'})
124
125    user.admin = True
126    db.session.commit()
127
128    return jsonify({'message' : 'The user has been promoted!'})
129
130@app.route('/user/<public_id>', methods=['DELETE'])
131@token_required
132def delete_user(current_user, public_id):
133    if not current_user.admin:
134        return jsonify({'message' : 'Cannot perform that function!'})
135
136    user = User.query.filter_by(public_id=public_id).first()
137
138    if not user:
139        return jsonify({'message' : 'No user found!'})
140
141    db.session.delete(user)
142    db.session.commit()
143
144    return jsonify({'message' : 'The user has been deleted!'})
145
146@app.route('/login')
147def login():
148    auth = request.authorization
149
150    if not auth or not auth.username or not auth.password:
151        return make_response('Could not verify', 401, {'WWW-Authenticate' : 'Basic realm="Login required!"'})
152
153    user = User.query.filter_by(name=auth.username).first()
154
155    if not user:
156        return make_response('Could not verify', 401, {'WWW-Authenticate' : 'Basic realm="Login required!"'})
157
158    if check_password_hash(user.password, auth.password):
159        token = jwt.encode({'public_id' : user.public_id, 'exp' : datetime.datetime.utcnow() + datetime.timedelta(minutes=30)}, app.config['SECRET_KEY'])
160
161        return jsonify({'token' : token.decode('UTF-8')})
162
163    return make_response('Could not verify', 401, {'WWW-Authenticate' : 'Basic realm="Login required!"'})
164
165@app.route('/todo', methods=['GET'])
166@token_required
167def get_all_todos(current_user):
168    todos = Todo.query.filter_by(user_id=current_user.id).all()
169
170    output = []
171
172    for todo in todos:
173        todo_data = {}
174        todo_data['id'] = todo.id
175        todo_data['text'] = todo.text
176        todo_data['complete'] = todo.complete
177        output.append(todo_data)
178
179    return jsonify({'todos' : output})
180
181@app.route('/todo/<todo_id>', methods=['GET'])
182@token_required
183def get_one_todo(current_user, todo_id):
184    todo = Todo.query.filter_by(id=todo_id, user_id=current_user.id).first()
185
186    if not todo:
187        return jsonify({'message' : 'No todo found!'})
188
189    todo_data = {}
190    todo_data['id'] = todo.id
191    todo_data['text'] = todo.text
192    todo_data['complete'] = todo.complete
193
194    return jsonify(todo_data)
195
196@app.route('/todo', methods=['POST'])
197@token_required
198def create_todo(current_user):
199    data = request.get_json()
200
201    new_todo = Todo(text=data['text'], complete=False, user_id=current_user.id)
202    db.session.add(new_todo)
203    db.session.commit()
204
205    return jsonify({'message' : "Todo created!"})
206
207@app.route('/todo/<todo_id>', methods=['PUT'])
208@token_required
209def complete_todo(current_user, todo_id):
210    todo = Todo.query.filter_by(id=todo_id, user_id=current_user.id).first()
211
212    if not todo:
213        return jsonify({'message' : 'No todo found!'})
214
215    todo.complete = True
216    db.session.commit()
217
218    return jsonify({'message' : 'Todo item has been completed!'})
219
220@app.route('/todo/<todo_id>', methods=['DELETE'])
221@token_required
222def delete_todo(current_user, todo_id):
223    todo = Todo.query.filter_by(id=todo_id, user_id=current_user.id).first()
224
225    if not todo:
226        return jsonify({'message' : 'No todo found!'})
227
228    db.session.delete(todo)
229    db.session.commit()
230
231    return jsonify({'message' : 'Todo item deleted!'})
232
233if __name__ == '__main__':
234    app.run(debug=True)