Flask CRUD with JWT Auth

 1 class Config(object):
 2     DEBUG = True
 3     TESTING = False
 4     SQLALCHEMY_TRACK_MODIFICATIONS = False
 5 
 6 
 7 class ProductionConfig(Config):
 8     SQLALCHEMY_DATABASE_URI = "mysql+pymysql://<db_url>:<port>/<db_name>"
 9     SQLALCHEMY_ECHO = False
10     JWT_SECRET_KEY = ‘JWT-SECRET‘
11     SECRET_KEY = ‘SECRET-KEY‘
12     SECURITY_PASSWORD_SALT = ‘SECRET-KEY-PASSWORD‘
13 
14 
15 class TestingConfig(Config):
16     TESTING = True
17     SQLALCHEMY_DATABASE_URI = "mysql+pymysql://<db_url>:<port>/<db_name>"
18     SQLALCHEMY_ECHO = False
19     JWT_SECRET_KEY = ‘JWT-SECRET‘
20     SECRET_KEY = ‘SECRET-KEY‘
21     SECURITY_PASSWORD_SALT = ‘SECRET-KEY-PASSWORD‘
22 
23 
24 class DevelopmentConfig(Config):
25     DEBUG = True
26     SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:123456@localhost:3306/test"
27     SQLALCHEMY_ECHO = False
28     JWT_SECRET_KEY = ‘JWT-SECRET‘
29     SECRET_KEY = ‘SECRET-KEY‘
30     SECURITY_PASSWORD_SALT = ‘SECRET-KEY-PASSWORD‘

1 from flask_sqlalchemy import SQLAlchemy
2 
3 db = SQLAlchemy()

 1 from flask import make_response, jsonify
 2 
 3 INVALID_FIELD_NAME_SENT_422 = {
 4     "http_code": 422,
 5     "code": "invalidField",
 6     "message": "Invalid fields found"
 7 }
 8 
 9 INVALID_INPUT_422 = {
10     "http_code": 422,
11     "code": "invalidInput",
12     "message": "Invalid input"
13 }
14 
15 MISSING_PARAMETERS_422 = {
16     "http_code": 422,
17     "code": "missingParameter",
18     "message": "Missing parameters."
19 }
20 
21 BAD_REQUEST_400 = {
22     "http_code": 400,
23     "code": "badRequest",
24     "message": "Bad request"
25 }
26 
27 SERVER_ERROR_500 = {
28     "http_code": 500,
29     "code": "serverError",
30     "message": "Server error"
31 }
32 
33 SERVER_ERROR_404 = {
34     "http_code": 404,
35     "code": "notFound",
36     "message": "Resource not found"
37 }
38 
39 FORBIDDEN_403 = {
40     "http_code": 403,
41     "code": "notAuthorized",
42     "message": "You are not authorised to execute this."
43 }
44 UNAUTHORIZED_401 = {
45     "http_code": 401,
46     "code": "notAuthorized",
47     "message": "Invalid authentication."
48 }
49 
50 NOT_FOUND_HANDLER_404 = {
51     "http_code": 404,
52     "code": "notFound",
53     "message": "route not found"
54 }
55 
56 SUCCESS_200 = {
57     ‘http_code‘: 200,
58     ‘code‘: ‘success‘
59 }
60 
61 SUCCESS_201 = {
62     ‘http_code‘: 201,
63     ‘code‘: ‘success‘,
64     ‘message‘: ‘resource has been created‘
65 }
66 
67 SUCCESS_204 = {
68     ‘http_code‘: 204,
69     ‘code‘: ‘success‘,
70     ‘message‘: ‘no data has been returned‘
71 }
72 
73 
74 def response_with(response, value=None, message=None, error=None, headers={}, pagination=None):
75     result = {}
76     if value is not None:
77         result.update(value)
78 
79     if response.get(‘message‘, None) is not None:
80         result.update({‘message‘: response[‘message‘]})
81 
82     result.update({‘code‘: response[‘code‘]})
83 
84     if error is not None:
85         result.update({‘errors‘: error})
86 
87     if pagination is not None:
88         result.update({‘pagination‘: pagination})
89 
90     headers.update({‘Access-Control-Allow-Origin‘: ‘*‘})
91     headers.update({‘server‘: ‘Flask REST API‘})
92 
93     return make_response(jsonify(result), response[‘http_code‘], headers)

 1 from marshmallow import fields
 2 from marshmallow_sqlalchemy import ModelSchema
 3 from passlib.hash import pbkdf2_sha256 as sha256
 4 
 5 from api.utils.database_util import db
 6 
 7 
 8 class User(db.Model):
 9     __tablename__ = ‘users‘
10 
11     id = db.Column(db.Integer, primary_key=True)
12     username = db.Column(db.String(120), unique=True, nullable=False)
13     password = db.Column(db.String(120), nullable=False)
14     created_time = db.Column(db.DateTime, server_default=db.func.now())
15     updated_time = db.Column(db.DateTime, server_default=db.func.now())
16 
17     def create(self):
18         db.session.add(self)
19         db.session.commit()
20         return self
21 
22     @classmethod
23     def find_by_email(cls, email):
24         return cls.query.filter_by(email=email).first()
25 
26     @classmethod
27     def find_by_username(cls, username):
28         return cls.query.filter_by(username=username).first()
29 
30     @staticmethod
31     def generate_hash(password):
32         return sha256.hash(password)
33 
34     @staticmethod
35     def verify_hash(password, hash):
36         return sha256.verify(password, hash)
37 
38 
39 class UserSchema(ModelSchema):
40     class Meta(ModelSchema.Meta):
41         model = User
42         sqla_session = db.session
43 
44     id = fields.Number(dump_only=True)
45     username = fields.String(required=True)
46     created_time = fields.String(dump_only=True)
47     updated_time = fields.String(dump_only=True)

 1 from marshmallow import fields
 2 from marshmallow_sqlalchemy import ModelSchema
 3 
 4 from api.models.books import BookSchema
 5 from api.utils.database_util import db
 6 
 7 
 8 class Author(db.Model):
 9     __tablename__ = ‘authors‘
10 
11     id = db.Column(db.Integer, primary_key=True, autoincrement=True)
12     first_name = db.Column(db.String(20))
13     last_name = db.Column(db.String(20))
14     created_time = db.Column(db.DateTime, server_default=db.func.now())
15     updated_time = db.Column(db.DateTime, server_default=db.func.now())
16     books = db.relationship(‘Book‘, backref=‘Author‘, cascade="all, delete-orphan")
17 
18     def __init__(self, first_name, last_name, books=[]):
19         self.first_name = first_name
20         self.last_name = last_name
21         self.books = books
22 
23     def create(self):
24         db.session.add(self)
25         db.session.commit()
26         return self
27 
28 
29 class AuthorSchema(ModelSchema):
30     class Meta(ModelSchema.Meta):
31         model = Author
32         sqla_session = db.session
33 
34     id = fields.Number(dump_only=True)
35     first_name = fields.String(required=True)
36     last_name = fields.String(required=True)
37     created_time = fields.String(dump_only=True)
38     updated_time = fields.String(dump_only=True)
39     books = fields.Nested(BookSchema, many=True, only=[‘title‘, ‘year‘, ‘id‘])

 1 from marshmallow import fields
 2 from marshmallow_sqlalchemy import ModelSchema
 3 
 4 from api.utils.database_util import db
 5 
 6 
 7 class Book(db.Model):
 8     __tablename__ = ‘books‘
 9 
10     id = db.Column(db.Integer, primary_key=True, autoincrement=True)
11     title = db.Column(db.String(50))
12     year = db.Column(db.Integer)
13     author_id = db.Column(db.Integer, db.ForeignKey(‘authors.id‘), nullable=False)
14     created_time = db.Column(db.DateTime, server_default=db.func.now())
15     updated_time = db.Column(db.DateTime, server_default=db.func.now())
16 
17     def __init__(self, title, year, author_id=None):
18         self.title = title
19         self.year = year
20         self.author_id = author_id
21 
22     def create(self):
23         db.session.add(self)
24         db.session.commit()
25         return self
26 
27 
28 class BookSchema(ModelSchema):
29     class Meta(ModelSchema.Meta):
30         model = Book
31         sqla_session = db.session
32 
33     id = fields.Number(dump_only=True)
34     title = fields.String(required=True)
35     year = fields.Integer(required=True)
36     created_time = fields.String(dump_only=True)
37     updated_time = fields.String(dump_only=True)
38     author_id = fields.Integer()

 1 from flask import Blueprint
 2 from flask import request
 3 from flask_jwt_extended import create_access_token
 4 
 5 import api.utils.response_util as resp
 6 from api.models.users import User, UserSchema
 7 from api.utils.response_util import response_with
 8 
 9 user_routes = Blueprint("user_routes", __name__)
10 
11 
12 @user_routes.route(‘/‘, methods=[‘POST‘])
13 def create_user():
14     try:
15         data = request.get_json()
16         data[‘password‘] = User.generate_hash(data[‘password‘])
17         user_schema = UserSchema()
18         user = user_schema.load(data)
19         result = user_schema.dump(user.create())
20         print(result)
21         return response_with(resp.SUCCESS_201)
22     except Exception as e:
23         print(e)
24         return response_with(resp.INVALID_INPUT_422)
25 
26 
27 @user_routes.route(‘/login‘, methods=[‘POST‘])
28 def authenticate_user():
29     try:
30         data = request.get_json()
31         current_user = User.find_by_username(data[‘username‘])
32         if not current_user:
33             return response_with(resp.SERVER_ERROR_404)
34         if User.verify_hash(data[‘password‘], current_user.password):
35             access_token = create_access_token(identity=data[‘username‘])
36             return response_with(resp.SUCCESS_201, value={‘message‘: ‘Logged in as {}‘.format(current_user.username),
37                                                           "access_token": access_token})
38         else:
39             return response_with(resp.UNAUTHORIZED_401)
40     except Exception as e:
41         print(e)
42         return response_with(resp.INVALID_INPUT_422)

 1 from flask import Blueprint, request
 2 from flask_jwt_extended import jwt_required
 3 from datetime import datetime
 4 
 5 from api.models.authors import Author, AuthorSchema
 6 from api.utils import response_util as resp
 7 from api.utils.database_util import db
 8 from api.utils.response_util import response_with
 9 
10 author_routes = Blueprint("author_routes", __name__)
11 
12 
13 @author_routes.route(‘/‘, methods=[‘POST‘])
14 @jwt_required
15 def create_author():
16     try:
17         data = request.get_json()
18         author_schema = AuthorSchema()
19         author = author_schema.load(data)
20         result = author_schema.dump(author.create())
21         return response_with(resp.SUCCESS_201, value={"author": result})
22     except Exception as e:
23         print(e)
24         return response_with(resp.INVALID_INPUT_422)
25 
26 
27 @author_routes.route(‘/‘, methods=[‘GET‘])
28 def get_author_list():
29     authors = Author.query.all()
30     author_schema = AuthorSchema(many=True, only=[‘first_name‘, ‘last_name‘, ‘id‘])
31     result = author_schema.dump(authors)
32     return response_with(resp.SUCCESS_200, value={"authors": result})
33 
34 
35 @author_routes.route(‘/<int:author_id>‘, methods=[‘GET‘])
36 def get_author_detail(author_id):
37     author = Author.query.get_or_404(author_id)
38     author_schema = AuthorSchema()
39     result = author_schema.dump(author)
40     return response_with(resp.SUCCESS_200, value={"author": result})
41 
42 
43 @author_routes.route(‘/<int:id>‘, methods=[‘PUT‘])
44 @jwt_required
45 def update_author_detail(id):
46     data = request.get_json()
47     author = Author.query.get_or_404(id)
48     author.first_name = data[‘first_name‘]
49     author.last_name = data[‘last_name‘]
50     author.updated_time = datetime.now()
51     db.session.add(author)
52     db.session.commit()
53     author_schema = AuthorSchema()
54     result = author_schema.dump(author)
55     return response_with(resp.SUCCESS_200, value={"author": result})
56 
57 
58 @author_routes.route(‘/<int:id>‘, methods=[‘PATCH‘])
59 @jwt_required
60 def modify_author_detail(id):
61     data = request.get_json()
62     author = Author.query.get(id)
63     if data.get(‘first_name‘):
64         author.first_name = data[‘first_name‘]
65     if data.get(‘last_name‘):
66         author.last_name = data[‘last_name‘]
67     author.updated_time = datetime.now()
68     db.session.add(author)
69     db.session.commit()
70     author_schema = AuthorSchema()
71     result = author_schema.dump(author)
72     return response_with(resp.SUCCESS_200, value={"author": result})
73 
74 
75 @author_routes.route(‘/<int:id>‘, methods=[‘DELETE‘])
76 @jwt_required
77 def delete_author(id):
78     author = Author.query.get_or_404(id)
79     db.session.delete(author)
80     db.session.commit()
81     return response_with(resp.SUCCESS_204)

 1 from flask import Blueprint
 2 from flask import request
 3 from flask_jwt_extended import jwt_required
 4 from datetime import datetime
 5 
 6 from api.models.books import Book, BookSchema
 7 from api.utils import response_util as resp
 8 from api.utils.database_util import db
 9 from api.utils.response_util import response_with
10 
11 book_routes = Blueprint("book_routes", __name__)
12 
13 
14 @book_routes.route(‘/‘, methods=[‘GET‘])
15 def get_book_list():
16     books = Book.query.all()
17     book_schema = BookSchema(many=True, only=[‘author_id‘, ‘title‘, ‘year‘])
18     result = book_schema.dump(books)
19     return response_with(resp.SUCCESS_200, value={"books": result})
20 
21 
22 @book_routes.route(‘/<int:id>‘, methods=[‘GET‘])
23 def get_book_detail(id):
24     book = Book.query.get_or_404(id)
25     book_schema = BookSchema()
26     result = book_schema.dump(book)
27     return response_with(resp.SUCCESS_200, value={"books": result})
28 
29 
30 @book_routes.route(‘/‘, methods=[‘POST‘])
31 @jwt_required
32 def create_book():
33     try:
34         data = request.get_json()
35         book_schema = BookSchema()
36         book = book_schema.load(data)
37         result = book_schema.dump(book.create())
38         return response_with(resp.SUCCESS_201, value={"book": result})
39     except Exception as e:
40         print(e)
41         return response_with(resp.INVALID_INPUT_422)
42 
43 
44 @book_routes.route(‘/<int:id>‘, methods=[‘PUT‘])
45 @jwt_required
46 def update_book_detail(id):
47     data = request.get_json()
48     book = Book.query.get_or_404(id)
49     book.title = data[‘title‘]
50     book.year = data[‘year‘]
51     book.updated_time = datetime.now()
52     db.session.add(book)
53     db.session.commit()
54     book_schema = BookSchema()
55     result = book_schema.dump(book)
56     return response_with(resp.SUCCESS_200, value={"book": result})
57 
58 
59 @book_routes.route(‘/<int:id>‘, methods=[‘PATCH‘])
60 @jwt_required
61 def modify_book_detail(id):
62     data = request.get_json()
63     book = Book.query.get_or_404(id)
64     if data.get(‘title‘):
65         book.title = data[‘title‘]
66     if data.get(‘year‘):
67         book.year = data[‘year‘]
68     book.updated_time = datetime.now()
69     db.session.add(book)
70     db.session.commit()
71     book_schema = BookSchema()
72     result = book_schema.dump(book)
73     return response_with(resp.SUCCESS_200, value={"book": result})
74 
75 
76 @book_routes.route(‘/<int:id>‘, methods=[‘DELETE‘])
77 @jwt_required
78 def delete_book(id):
79     book = Book.query.get_or_404(id)
80     db.session.delete(book)
81     db.session.commit()
82     return response_with(resp.SUCCESS_204)

 1 import logging
 2 import os
 3 
 4 from flask import Flask
 5 from flask_jwt_extended import JWTManager
 6 
 7 import api.utils.response_util as resp
 8 from api.config.env_config import DevelopmentConfig, ProductionConfig, TestingConfig
 9 from api.routes.author_routes import author_routes
10 from api.routes.book_routes import book_routes
11 from api.routes.user_routes import user_routes
12 from api.utils.database_util import db
13 from api.utils.response_util import response_with
14 
15 app = Flask(__name__)
16 
17 if os.environ.get(‘WORK_ENV‘) == ‘PROD‘:
18     app_config = ProductionConfig
19 elif os.environ.get(‘WORK_ENV‘) == ‘TEST‘:
20     app_config = TestingConfig
21 else:
22     app_config = DevelopmentConfig
23 
24 app.config.from_object(app_config)
25 
26 db.init_app(app)
27 with app.app_context():
28     db.create_all()
29 app.register_blueprint(author_routes, url_prefix=‘/api/authors‘)
30 app.register_blueprint(book_routes, url_prefix=‘/api/books‘)
31 app.register_blueprint(user_routes, url_prefix=‘/api/users‘)
32 
33 
34 @app.after_request
35 def add_header(response):
36     logging.info(response)
37     return response
38 
39 
40 @app.errorhandler(400)
41 def bad_request(e):
42     logging.error(e)
43     return response_with(resp.BAD_REQUEST_400)
44 
45 
46 @app.errorhandler(500)
47 def server_error(e):
48     logging.error(e)
49     return response_with(resp.SERVER_ERROR_500)
50 
51 
52 @app.errorhandler(404)
53 def not_found(e):
54     logging.error(e)
55     return response_with(resp.SERVER_ERROR_404)
56 
57 
58 jwt = JWTManager(app)
59 db.init_app(app)
60 with app.app_context():
61     db.create_all()
62 
63 if __name__ == "__main__":
64     app.run(port=5000, host="0.0.0.0", use_reloader=False)

1 from main import app as application
2 
3 if __name__ == "__main__":
4     application.run()

 * Serving Flask app "main" (lazy loading)
 * Environment: production
   WARNING: This is a development server. Do not use it in a production deployment.
   Use a production WSGI server instead.
 * Debug mode: on
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 444-563-988

$ curl http://127.0.0.1:5000/api/users/ -d ‘{
        "username": "Flask",
        "password": "Python"
}‘ -X POST -H ‘Content-Type: application/json‘

{
  "code": "success",
  "message": "resource has been created"
}

$ curl http://127.0.0.1:5000/api/users/login -d ‘{
        "username": "Flask",
        "password": "Python"
}‘ -X POST -H ‘Content-Type: application/json‘

{
  "access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpYXQiOjE1Nzg0NzcwMDEsIm5iZiI6MTU3ODQ3NzAwMSwianRpIjoiNmE3MzczMjgtZWM3OS00ZTJhLTkyYWMtMzhhMWU2NWZlZWZlIiwiZXhwIjoxNTc4NDc3OTAxLCJpZGVudGl0eSI6IkZsYXNrIiwiZnJlc2giOmZhbHNlLCJ0eXBlIjoiYWNjZXNzIn0.awKg8wSSvJq-_P60_lAB-AgzdlSS_SX-voTbCXeUdMY",
  "code": "success",
  "message": "resource has been created"
}

$ curl http://127.0.0.1:5000/api/authors/ -d ‘{
        "first_name": "Hello",
        "last_name": "World"
}‘ -X POST -H ‘Content-Type: application/json‘ -H ‘Authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpYXQiOjE1Nzg0NzcwMDEsIm5iZiI6MTU3ODQ3NzAwMSwianRpIjoiNmE3MzczMjgtZWM3OS00ZTJhLTkyYWMtMzhhMWU2NWZlZWZlIiwiZXhwIjoxNTc4NDc3OTAxLCJpZGVudGl0eSI6IkZsYXNrIiwiZnJlc2giOmZhbHNlLCJ0eXBlIjoiYWNjZXNzIn0.awKg8wSSvJq-_P60_lAB-AgzdlSS_SX-voTbCXeUdMY‘

{
  "author": {
    "books": [],
    "created_time": "2020-01-08 09:54:52",
    "first_name": "Hello",
    "id": 10.0,
    "last_name": "World",
    "updated_time": "2020-01-08 09:54:52"
  },
  "code": "success",
  "message": "resource has been created"
}

$ curl http://127.0.0.1:5000/api/authors/10
{
  "author": {
    "books": [],
    "created_time": "2020-01-08 09:54:52",
    "first_name": "Hello",
    "id": 10.0,
    "last_name": "World",
    "updated_time": "2020-01-08 09:54:52"
  },
  "code": "success"
}