原创转载请注明出处：https://www.cnblogs.com/agilestyle/p/12167897.html
Project Directory
Note: 所有 __init__.py 都是空的
Config
env_config.py
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'
Utils
database_util.py
1 from flask_sqlalchemy import SQLAlchemy 2 3 db = SQLAlchemy()
response_util.py
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)
Models
user.py
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)
authors.py
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'])
books.py
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()
Routes
user_routes.py
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)
author_routes.py
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)
book_routes.py
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)
Main
main.py
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)
run.py
1 from main import app as application 2 3 if __name__ == "__main__": 4 application.run()
Test
启动 run.py
* 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
POST user create
$ 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" }
POST user login
$ 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"
}
POST author create
$ 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" }
Note: token取得是user login返回的token值
GET author
$ 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" }
Note: GET方法不需要验证JWT，其他路由测试以此类推。