python 烧瓶相关代码

Posted 2021-05-09

# 常用的命令


# 生成项目依赖的包
pip freeze > ./requirements.txt
# 安装txt文件中的包
pip install -r requirements.txt

# -*- coding: utf-8 -*-
"""
@zhuzhezhe create on 2018/1/23
copyright@2018
"""
from flask import Flask, g, request, abort, jsonify, url_for
from flask_httpauth import HTTPBasicAuth
from flask_sqlalchemy import SQLAlchemy
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer, SignatureExpired, BadSignature
from werkzeug.security import generate_password_hash, check_password_hash

app = Flask(__name__)
app.config['SECRET_KEY'] = 'the quick brown fox jumps over the lazy dog'
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///db.sqlite'
app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True
auth = HTTPBasicAuth()
db = SQLAlchemy(app)

"""数据库"""


class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(32), index=True)
    # 密码经过hash后才存入数据库
    password_hash = db.Column(db.String(64))

    # 设置密码不可读，同时将密码hash
    @property
    def password(self):
        raise AttributeError('password is not a readble attribute')

    @password.setter
    def password(self, password):
        self.password_hash = generate_password_hash(password)

    # 验证客户端发过来的密码是否与数据库中hash后的密码一致
    def verify_password(self, password):
        return check_password_hash(self.password_hash, password)

    # 根据用户id生成令牌，过期时间600秒
    def generate_auth_token(self, expiration=600):
        s = Serializer(app.config['SECRET_KEY'], expires_in=expiration)
        return s.dumps({'id': self.id})

    # 根据客户端发过来的token找出对应的用户id
    # 注意静态方法，因为只有解码令牌后才可能知道用户是谁
    @staticmethod
    def verify_auth_token(token):
        s = Serializer(app.config['SECRET_KEY'])
        try:
            data = s.loads(token)
        except SignatureExpired:
            return None
        except BadSignature:
            return None
        user = User.query.get(data['id'])
        return user


@auth.verify_password
def verify_password(username_or_token, password):
    user = User.verify_auth_token(username_or_token)
    # 先检查令牌
    if not user:
        user = User.query.filter_by(username_or_token=username_or_token).first()
        # 不是令牌检查密码
        if not user or not user.verify_password(password):
            return False
    # 设置全局变量g为当前user
    g.user = user
    return True


"""api接口"""


# 获取当前用户的token
# 要求用户必须先登录
@app.route('/api/v1.0/token')
@auth.login_required
def get_auth_token():
    # 为当前用户生成token
    token = g.user.generate_auth_token()
    return jsonify({'token': token, 'deration': 600})


@app.route('/api/v1.0/users/<int:id>', methods=['GET'])
def get_user(id):
    user = User.query.get(id)
    if not user:
        abort(400)
    return jsonify({'username': user.username})


@app.route('/api/v1.0/users', methods=['POST'])
def create_user():
    username = request.json.get('username')
    password = request.json.get('password')
    if username is None or password is None:
        abort(400)
    if User.query.filter_by(username=username).first() is not None:
        abort(400)
    user = User(username=username)
    user.password_hash(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/resource')
@auth.login_required
def get_resource():
    return jsonify({'data': 'Hello, %s!' % g.user.username})


if __name__ == '__main__':
    app.run(debug=True)