使用Flask-Restful实现API接口

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

1. 这部分是主要的代码，从flask_restful库中导入Api和Resouce类，并且创建Api对象，创建对象时需要传入Blueprint对象，即api_bp。
2. 实现一个接口时，需要创建一个类，并且该类要继承Resouce。
3. 在继承了Resouce的类中，post表示接收的是POST请求，get接收的是get请求....其他一样。
4. 使用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')
5. (4)中的访问完整链接是http://127.0.0.1:5000/api/useradd
6. 在定义接口时，返回的数据类型都应该是json格式
7. 网页访问时可能会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