1. 代码结构：
2. 具体代码
2.1 user.py
from ApiRESTful.extensions import db
from werkzeug.security import generate_password_hash, check_password_hash
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(64), unique=True, index=True)
# 将password设置为私有属性，并且重命名
_password = db.Column('password', db.String(128))
# 定义一个属性，默认是读取的操作，这里报错，意思是不可读
@property
def password(self):
raise AttributeError('password is not readable attribute')
# 定义上面那个password属性的可写属性，这里默认换算成哈希值，然后保存下来
@password.setter
def password(self, password):
self._password = generate_password_hash(password)
# 校验传入的密码和哈希值是否是一对儿
def verify_password(self, password):
return check_password_hash(self._password, password)
def __repr__(self):
return "<User {}>".format(self.username)
2.2 api_auth.py
Flask-Httpauth是用来验证用户的，但在这个部分中没有用到，之后会专门来写一个。
from flask_httpauth import HTTPBasicAuth
from flask import jsonify, app
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
from itsdangerous import SignatureExpired, BadSignature
from ApiRESTful.settings import DevelopConfig
from ApiRESTful.models.user import User
auth = HTTPBasicAuth()
@(FLASK)auth.error_handler
def unauthorized():
error_info = '{}'.format('Invalid credentials')
print('api.auth.unauthorized.error_info = ' + error_info)
response = jsonify({'error': error_info})
response.status_code = 403
print('api.auth.unauthorized.response = ' + str(response))
return response
def verify_password_for_token(username, password):
"""
验证输入的用户名和密码是否匹配
:param username: 用户名
:param password: 密码
:return: 匹配结果
"""
user = User.query.filter_by(username=username).first()
# 注意一下，以后判断是否为空，还是写成 user is None 而不要写成 not user
# 这样写更容易理解一些
if user is None or not user.verify_password(password):
# 结果不匹配
return False
return True
@auth.verify_password
def verify_password(username_or_token, password):
user = verify_auth_token(username_or_token)
if user is None:
return verify_password_for_token(username_or_token, password)
return user
def generator_auth_token(expiration=600):
s = Serializer(secret_key=DevelopConfig.SECRET_KEY, expires_in=expiration)
return s.dumps({'id': 1})
def verify_auth_token(token):
s = Serializer(DevelopConfig.SECRET_KEY)
try:
data = s.loads(token)
except SignatureExpired:
return None
except BadSignature:
return None
user = User.query.get(data.get('id'))
return user
2.3 errors.py
使用Blurprint的errorhandler来定义错误处理函数。
from flask import jsonify
from ApiRESTful.blueprints.api import api_bp
@api_bp.errorhandler(404)
def not_found(e):
print('api.errors.not_found ', e)
error_info = '{}'.format(e)
response = jsonify({'error': error_info})
response.status_code = 404
return response
@api_bp.errorhandler(403)
def fobidden(e):
print(e)
error_info = '{}'.format(e)
response = jsonify({'error': error_info})
response.status_code = 403
return response
2.4 api/__init__.py
在init.py中创建Blueprint对象api_bp
from flask import Blueprint
api_bp = Blueprint('api', __name__, url_prefix='/api')
from . import api_auth, api_user
2.5 api.user.py
- 这部分是主要的代码，从flask_restful库中导入Api和Resouce类，并且创建Api对象，创建对象时需要传入Blueprint对象，即api_bp。
- 实现一个接口时，需要创建一个类，并且该类要继承Resouce。
- 在继承了Resouce的类中，post表示接收的是POST请求，get接收的是get请求....其他一样。
- 使用
api_user.add_resource函数去申明接口的访问url，api_user是Api对象，例如
api_user.add_resource(UserAddApi, '/useradd', endpoint='useradd')
第一个参数是Resouce类，即上面定义的继承了Resouce的子类，在IDE显示警告，不用在意；第二个参数就是path了；第三个参数的endpoint是在代码内部访问的，跟Flask中一样，在一般情况下是不要写的，即访问方式url_for('api_user.useradd') - (4)中的访问完整链接是
http://127.0.0.1:5000/api/useradd - 在定义接口时，返回的数据类型都应该是json格式
- 网页访问时可能会post数据，数据类型是json格式的话，需要将它进行转换
json.loads(request.get_json())
import time
import json
from ApiRESTful.extensions import db
from flask_restful import Api, Resource
from flask import jsonify, request
from . import api_bp
from ApiRESTful.models.user import User
from ApiRESTful.blueprints.api.api_auth import auth, generator_auth_token, verify_auth_token
api_user = Api(api_bp)
class UserAddApi(Resource):
def post(self):
print('UserAddApi.post.url = ' + str(request.url))
# 如果传入进来的是json数据，那么需要转换成dict类型的数据
user_info = json.loads(request.get_json())
print('user_info.type =', type(user_info))
print('UserAddApi.post.user_info = ' + str(user_info))
try:
u = User(username=user_info['username'])
u.password = user_info['password']
db.session.add(u)
db.session.commit()
except:
print("{} User add: {} failure...".format(time.strftime("%Y-%m-%d %H:%M:%S"), user_info['username']))
db.session.rollback()
return False
else:
print("{} User add: {} success...".format(time.strftime("%Y-%m-%d %H:%M:%S"), user_info['username']))
return True
finally:
db.session.close()
class UserVerifyApi(Resource):
def post(self):
print('UserVerifyApi.post.url = ' + str(request.url))
user_info = json.loads(request.get_json())
try:
u = User.query.filter_by(username=user_info['username']).first()
if u is None or u.verify_password(user_info['password']) is False:
print("{} User query: {} failure...".format(time.strftime("%Y-%m-%d %H:%M:%S"), user_info['username']))
return False
except:
print("{} User query: {} failure...".format(time.strftime("%Y-%m-%d %H:%M:%S"), user_info['username']))
return False
else:
print("{} User query: {} success...".format(time.strftime("%Y-%m-%d %H:%M:%S"), user_info['username']))
return True
finally:
db.session.close()
class UserTokenApi(Resource):
# @auth.login_required
def get(self):
print('UserTokenApi.get.url = ' + str(request.url))
token = generator_auth_token(expiration=600)
return jsonify({'token': token.decode('ascii')})
api_user.add_resource(UserAddApi, '/useradd', endpoint='useradd')
api_user.add_resource(UserVerifyApi, '/userverify', endpoint='userverify')
api_user.add_resource(UserTokenApi, '/usertoken', endpoint='usertoken')
2.6 extensions.py
该类都会用来放第三方插件的对象
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
2.7 settings.py
配置类，会根据不同的场景配置不同的属性，这个只是demo，所以就不那么严格了。
import os
basedir = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
class DevelopConfig(object):
SECRET_KEY = 'a random string'
SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://root:root@localhost:3306/api'
SQLALCHEMY_COMMIT_ON_TEARDOWN = True
SQLALCHEMY_TRACK_MODIFICATIONS = False
DEBUG = True
2.8 ApiRESTful/__init__.py
创建Flask对象，以及做一些初始化操作，例如导入第三方插件的对象，然后调用init_app(app)等操作。
from flask import Flask
from ApiRESTful.settings import DevelopConfig
from ApiRESTful.extensions import db
from ApiRESTful.blueprints.api import api_bp
def register_blueprints(_app):
_app.register_blueprint(api_bp)
def register_extensions(_app):
db.init_app(_app)
init_db(_app)
def init_db(_app: Flask):
_app.app_context().push()
db.drop_all()
db.create_all()
db.session.commit()
def _create_app():
_app = Flask(__name__)
_app.config.from_object(DevelopConfig)
register_blueprints(_app)
register_extensions(_app)
return _app
app = _create_app()
2.9 manage.py
程序入口，执行manage.py就可以启动web应用了。
from flask_script import Manager, Server, Shell
from ApiRESTful import app
from ApiRESTful.extensions import db
from ApiRESTful.models.user import User
manager = Manager(app)
def make_shell_context():
return dict(app=app, db=db, User=User)
manager.add_command('runserver', Server(host='127.0.0.1', port=5000, use_debugger=True, use_reloader=True))
manager.add_command('shell', Shell(make_context=make_shell_context))
if __name__ == '__main__':
manager.run(default_command='runserver')
3.测试
测试代码
import requests
resp = requests.post(url='http://127.0.0.1:5000/api/useradd',
json="{\"username\": \"chentao\", \"password\": \"123456\"}")
print(resp)
r2 = requests.get(url='http://127.0.0.1:5000/api/usertoken')
print(r2)
print(r2.json())
r3 = requests.post(url='http://127.0.0.1:5000/api/userverify',
json="{\"username\": \"chentao\", \"password\": \"123456\"}")
print(r3.json())
print(r3)
4.参考
https://www.jianshu.com/p/81cd461c7e8f
https://flask-httpauth.readthedocs.io/en/latest/
http://www.pythondoc.com/Flask-RESTful/quickstart.html
