flask restful-api实现及基于flask-httpauth实现基础权限管控

Posted 2021-02-19

本系列教程分为四个阶段

1.flask restful web service

2.flask restful api

3.flask httpauth实现权限管控

4.uwsgi管理flask应用

?
前两篇文章中，我们学习了基础的web services和restful api的开发实现，在实现接口开发后，我们一般需要对接口进行加密认证，避免接口外泄导致的数据丢失等问题。
?
在原文中代码分为几部分说明，并不直观，而且影响连续性的学习，所以这里笔者直接通过整段代码的形式进行学习，并对代码中的关键点进行拆分讲解。
?
下面是实现httpauth及token认证的代码：

import time

from flask import Flask, jsonify, url_for, request, g
from flask_restful import Api, abort
# flask SQLAlchemy模块，实现ORM框架操作
from flask_sqlalchemy import SQLAlchemy
# flask自带httpauth模块，BasicAuth和tokenAuth子模块
from flask_httpauth import HTTPBasicAuth, HTTPTokenAuth
# 密码散列加密模块
from passlib.apps import custom_app_context as pwd_context
# jwt-- token生成模块
import jwt
# werkzeug密码加密模块，有兴趣的可以自行了解
# from werkzeug.security import generate_password_hash, check_password_hash

app = Flask(__name__)
api = Api(app)

# 关于flask SQLAlchemy实现ORM框架操作的一些基础配置
app.config[‘SECRET_KEY‘] = ‘dqwecf29vbneuirjnf2i3n0f2i302n‘
# 数据库连接信息配置
app.config[‘SQLALCHEMY_DATABASE_URI‘] = ‘sqlite:///db.sqlite‘
# 是否自动提交sql执行
app.config[‘SQLALCHEMY_COMMIT_ON_TEARDOWN‘] = True
# 是否显示修改回执
app.config[‘SQLALCHEMY_TRACK_MODIFICATIONS‘] = True

# SQLAlchemy实例化
db = SQLAlchemy(app)
# 引入基础认证、token认证模块
auth = HTTPBasicAuth()
token_m = HTTPTokenAuth()

class User(db.Model):
    #表名
    __tablename__ = ‘users‘
    #表字段名对应赋值属性
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(32), index=True)
    password_hash = db.Column(db.String(64))
    # print(id, username, password_hash)

    # def __init__(self, username, password_hash):
    #     self.username = username
    #     self.password_hash = password_hash

    #生成密码散列方法
    def hash_password(self, password):
        self.password_hash = pwd_context.encrypt(password)

    #校验密码方法
    def verify_password(self, password):
        return pwd_context.verify(password, self.password_hash)

    #生成token方法
    def generate_auth_token(self):
        # print(time.time())
        print("self id ", self.id)
        print(self.username)
        print(self.password_hash)
        return jwt.encode(
            {‘id‘: self.id, ‘exp‘: time.time() + 600},
            app.config[‘SECRET_KEY‘], algorithm=‘HS256‘)

    #校验token方法
    @staticmethod
    def verify_auth_token(token):
        try:
            data = jwt.decode(token, app.config[‘SECRET_KEY‘],
                              algorithms=[‘HS256‘])
            print(data)
        except Exception as e:
            print(e)
            return False
        user = User.query.get(data[‘id‘])
        print("user", user)
        return user

# 在初次请求时，进行数据的表创建
# 实际开发中，此处一般会在数据库中已经完成，研发人员只需要关注表结构，并进行对应的字段操作即可
@app.before_first_request
def create_db():
    #删除
    db.drop_all()
    #创建
    db.create_all()

# 根据密码登录认证装饰器进行请求携带的密码信息校验
@auth.verify_password
def verify_password(username_or_token, password):
    user = User.query.filter_by(username=username_or_token).first()
    if not user or not user.verify_password(password):
        return False
    g.user = user
    return True

# 根据token登录认证装饰器进行请求携带的token校验
@token_m.verify_token
def verify_token(token):
    user = User.verify_auth_token(token)
    g.user = user
    if not user:
        return False
    return True

# 获取用户访问信息，并根据判断信息，添加新用户
@app.route(‘/api/users‘, methods=[‘POST‘])
def new_user():
    username = request.json.get(‘username‘)
    password = request.json.get(‘password‘)
    if username is None or password is None:
        abort(400)  # missing arguments
    if User.query.filter_by(username=username).first() is not None:
        print(‘existing user‘)
        abort(400)  # existing user
    user = User(username=username)
    user.hash_password(password)
    db.session.add(user)
    db.session.commit()
    return (jsonify({‘username‘: user.username}), 201,
            {‘Location‘: url_for(‘get_user‘, id=user.id, _external=True)})

# 查询指定用户方法
@app.route(‘/api/users/<int:id>‘)
def get_user(id):
    user = User.query.get(id)
    if not user:
        abort(400)
    return jsonify({‘username‘: user.username})

# 查询所有用户方法
@app.route(‘/api/users‘, methods=[‘GET‘])
def get_all_user():
    users = User.query.all()
    if not users:
        abort(400)
    return jsonify(["{0}, {1}".format(user.username, user.password_hash) for user in users])

# 基于密码生成token方法
@app.route(‘/api/token‘)
# 密码登录验证装饰器
@auth.login_required
def get_auth_token():
    token = g.user.generate_auth_token()
    return jsonify({‘token‘: token.decode(‘ascii‘), ‘duration‘: 600})

# token认证方法
@app.route(‘/api/resource‘)
# token登录验证装饰器
@token_m.login_required
def get_resource():
    return jsonify({‘data‘: ‘Hello, %s!‘ % g.user.username})

@app.errorhandler(400)
def error(e):
    print(e)
    return ‘错误请求‘, 400

if __name__ == ‘__main__‘:
    app.run(‘0.0.0.0‘, 8080)

?
以上代码块中，读者可直接拷贝到IDEA中运行查看效果即可。
对应的需要了解的有，在ORM框架中，使用方法调用代替原生sql执行，如：

查询
User.query.filter_by(username=username).first()
增加
user = User(username=username)
user.hash_password(password)
db.session.add(user)
db.session.commit()

关于具体的基于SQLAlchemy实现的ORM框架使用方法，请各位自行查询资料了解学习。
代码块中已经对相应的模块进行说明，在实际调用中代码名称已经是自解释的，如若不了解，可以再搜索相关资料。
?
本文仅对flask的auth实现进行演示，不再过多的阐述，成年人的学习不应是喂什么吃什么，而是自己动手，丰衣足食。
?
感谢观看。