Last active
March 29, 2018 03:23
-
-
Save donrokzon/2405fe27245a036186f35142b48b84d0 to your computer and use it in GitHub Desktop.
Flask rest
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from flask import Flask, request, jsonify, make_response | |
| from flask_sqlalchemy import SQLAlchemy | |
| import uuid | |
| from werkzeug.security import generate_password_hash, check_password_hash | |
| import jwt | |
| import datetime | |
| from functools import wraps | |
| app = Flask(__name__) | |
| app.config['SECRET_KEY'] = 'thisissecret' | |
| app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:root@localhost/flask_db' | |
| db = SQLAlchemy(app) | |
| class User(db.Model): | |
| id = db.Column(db.Integer, primary_key=True) | |
| public_id = db.Column(db.String(50), unique=True) | |
| name = db.Column(db.String(50)) | |
| password = db.Column(db.String(80)) | |
| admin = db.Column(db.Boolean) | |
| class Todo(db.Model): | |
| id = db.Column(db.Integer, primary_key=True) | |
| text = db.Column(db.String(50)) | |
| complete = db.Column(db.Boolean) | |
| user_id = db.Column(db.Integer) | |
| def token_required(f): | |
| @wraps(f) | |
| def decorated(*args, **kwargs): | |
| token = None | |
| if 'x-access-token' in request.headers: | |
| token = request.headers['x-access-token'] | |
| if not token: | |
| return jsonify({"message": "Token is missing"}), 401 | |
| try: | |
| data = jwt.decode(token, app.config['SECRET_KEY']) | |
| current_user = User.query.filter_by(public_id=data['public_id']).first() | |
| except: | |
| return jsonify({'message': 'Token is invalid'}), 401 | |
| return f(current_user, *args, **kwargs) | |
| return decorated | |
| @app.route('/login', methods=['GET']) | |
| def login(): | |
| auth = request.authorization | |
| if not auth or not auth.username or not auth.password: | |
| return make_response('Could not verify', 401, {'WWW-Authenticate': 'Basic realm="Login required!"'}) | |
| user = User.query.filter_by(name=auth.username).first() | |
| if not user: | |
| return make_response('Could not verify', 401, {'WWW-Authenticate': 'Basic realm="Login required!"'}) | |
| if check_password_hash(user.password, auth.password): | |
| token = jwt.encode( | |
| {'public_id': user.public_id, 'exp': datetime.datetime.utcnow() + datetime.timedelta(minutes=30)}, | |
| app.config['SECRET_KEY']) | |
| return jsonify({'token': token.decode('UTF-8')}) | |
| return make_response('Could not verify', 401, {'WWW-Authenticate': 'Basic realm="Login required!"'}) | |
| @app.route('/user', methods=['GET']) | |
| @token_required | |
| def get_all_user(current_user): | |
| if not current_user.admin: | |
| return jsonify({"message": "Cannot perform that action!"}) | |
| users = User.query.all() | |
| output = [] | |
| for user in users: | |
| user_data = {} | |
| user_data['public_id'] = user.public_id | |
| user_data['name'] = user.name | |
| user_data['password'] = user.password | |
| user_data['admin'] = user.admin | |
| output.append(user_data) | |
| return jsonify({'users': output}) | |
| @app.route('/user/<public_id>', methods=['GET']) | |
| @token_required | |
| def get_one_user(current_user, public_id): | |
| user = User.query.filter_by(public_id=public_id).first() | |
| if not user: | |
| return jsonify({"message": "No user found"}) | |
| user_data = {} | |
| user_data['public_id'] = user.public_id | |
| user_data['name'] = user.name | |
| user_data['password'] = user.password | |
| user_data['admin'] = user.admin | |
| return jsonify({'users': user_data}) | |
| @app.route('/user', methods=['POST']) | |
| @token_required | |
| def create_user(current_user): | |
| data = request.get_json() | |
| hashed_password = generate_password_hash(data['password'], method='sha256') | |
| new_user = User(public_id=str(uuid.uuid4()), name=data['name'], password=hashed_password, admin=False) | |
| db.session.add(new_user) | |
| db.session.commit() | |
| return jsonify({'message': 'New user created'}) | |
| @app.route('/user/<public_id>', methods=['PUT']) | |
| @token_required | |
| def promote_user(current_user, public_id): | |
| user = User.query.filter_by(public_id=public_id).first() | |
| if not user: | |
| return jsonify({"message": "No user found"}) | |
| user.admin = True | |
| db.session.commit() | |
| return jsonify({"message": "user has been promoted"}) | |
| @app.route('/user/<public_id>', methods=['DELETE']) | |
| @token_required | |
| def delete_user(current_user, public_id): | |
| user = User.query.filter_by(public_id=public_id).first() | |
| if not user: | |
| return jsonify({"message": "No user found"}) | |
| db.session.delete(user) | |
| db.session.commit() | |
| return jsonify({"message": "The user " + user.name + " has been deleted"}) | |
| if __name__ == '__main__': | |
| app.run(debug=True) | |
| db.create_all() | |
| db.session.commit() | |
| admin = User('admin', 'admin@example.com') | |
| guest = User('guest', 'guest@example.com') | |
| db.session.add(admin) | |
| db.session.add(guest) | |
| db.session.commit() | |
| users = User.query.all() | |
| print users | |
| apt-get install python-mysqldb | |
| windows pip install mysqlclient |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment