Flask Vue.js全栈开发|第3章:Flask设计User用户相关API

  • 原创
  • Madman
  • /
  • 2018-11-02 10:06
  • /
  • 0
  • 134 次阅读

flask vuejs 全栈开发-min.png

Synopsis: Flask 后端针对 "用户资源" 提供部分 RESTful API,基于 token 认证,目前支持添加用户、查看单个或多个用户、修改用户,使用 HTTPie 或 Postman 测试 API 通过。下一篇将在前端使用这些 API 实现用户注册、登录与退出功能

代码已上传到 https://github.com/wangy8961/flask-vuejs-madblog/tree/v0.3 ,欢迎star

1. git

1.1 从 Github 拉取最新代码

$ git remote -v
origin  git@github.com:wangy8961/flask-vuejs-madblog.git (fetch)
origin  git@github.com:wangy8961/flask-vuejs-madblog.git (push)

$ git fetch
或者,拉取指定的远程主机上的分支,比如 origin 上的 master 分支
$ git fetch origin master

1.2 创建 dev 分支

$ git checkout -b dev
$ git branch

2. 数据库

2.1 ORM: SQLAlchemy

安装 Flask-SQLAlchemy 插件,还有数据表结构有变化后进行迁移的 Flask-Migrate 插件

(venv) D:\python-code\flask-vuejs-madblog\back-end>pip install flask-sqlalchemy flask-migrate
(venv) D:\python-code\flask-vuejs-madblog\back-end>pip freeze > requirements.txt

修改配置文件 back-end/config.py,默认使用 SQLite 数据库:

import os
from dotenv import load_dotenv

basedir = os.path.abspath(os.path.dirname(__file__))
load_dotenv(os.path.join(basedir, '.env'))


class Config(object):
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
        'sqlite:///' + os.path.join(basedir, 'app.db')
    SQLALCHEMY_TRACK_MODIFICATIONS = False

修改 app/__init__.py,引入并初始化插件:

from flask import Flask
from flask_cors import CORS
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from config import Config


# Flask-SQLAlchemy plugin
db = SQLAlchemy()
# # Flask-Migrate plugin
migrate = Migrate()


def create_app(config_class=Config):
    app = Flask(__name__)
    app.config.from_object(config_class)

    # Enable CORS
    CORS(app)
    # Init Flask-SQLAlchemy
    db.init_app(app)
    # Init Flask-Migrate
    migrate.init_app(app, db)

    # 注册 blueprint
    from app.api import bp as api_bp
    app.register_blueprint(api_bp, url_prefix='/api')

    return app

2.2 定义 User 用户数据模型

创建 app/models.py

from app import db


class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), index=True, unique=True)
    email = db.Column(db.String(120), index=True, unique=True)
    password_hash = db.Column(db.String(128))  # 不保存原始密码

    def __repr__(self):
        return '<User {}>'.format(self.username)

修改 app/__init__.py,在文件末尾添加:

from app import models

2.3 第一次数据库迁移

(1) 创建迁移存储库

(venv) D:\python-code\flask-vuejs-madblog\back-end>flask db init

(2) 生成迁移脚本

(venv) D:\python-code\flask-vuejs-madblog\back-end>flask db migrate -m "add users table"

(3) 将迁移脚本应用到数据库中

(venv) D:\python-code\flask-vuejs-madblog\back-end>flask db upgrade

说明: flask db downgrade 命令可以回滚上次的迁移

2.4 存储用户密码的 hash 值

使用 werkzeug.security 库的 generate_password_hashcheck_password_hash 来创建哈希密码和验证密码的hash是否一致:

(venv) D:\python-code\flask-vuejs-madblog\back-end>python
Python 3.6.1 (v3.6.1:69c0db5, Mar 21 2017, 17:54:52) [MSC v.1900 32 bit (Intel)] on win32
Type "help", "copyright", "credits" or "license" for more information.
>>> from werkzeug.security import generate_password_hash, check_password_hash
>>> hash = generate_password_hash('foobar')
>>> hash
'pbkdf2:sha256:50000$Z39YZhom$2a59a7a0edf67db5e29632134cb1fbbfec55077a262c659de662dbe5de623329'
>>> check_password_hash(hash, 'foobar')
True
>>> check_password_hash(hash, 'barfoo')
False
>>>

更新 User 数据模型:

from werkzeug.security import generate_password_hash, check_password_hash
from app import db


class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), index=True, unique=True)
    email = db.Column(db.String(120), index=True, unique=True)
    password_hash = db.Column(db.String(128))  # 不保存原始密码

    def __repr__(self):
        return '<User {}>'.format(self.username)

    def set_password(self, password):
        self.password_hash = generate_password_hash(password)

    def check_password(self, password):
        return check_password_hash(self.password_hash, password)

配置 Flask Shell 上下文环境:

flask shell 命令是 flask 命令集中的另一个非常有用的工具,它是继 flask run 之后被实现的第二个 "核心" 命令,其目的是启动一个Python解释器包含应用的上下文

(venv) D:\python-code\flask-vuejs-madblog\back-end>flask shell
Python 3.5.2 (v3.5.2:4def2a2901a5, Jun 25 2016, 22:18:55) [MSC v.1900 64 bit (AMD64)] on win32
App: app [production]
Instance: D:\python-code\flask-vuejs-madblog\back-end\instance
>>> app
<Flask 'app'>
>>> db
Traceback (most recent call last):
  File "<console>", line 1, in <module>
NameError: name 'db' is not defined

修改 back-end/madblog.py,添加一个方法:

from app import create_app, db
from app.models import User

app = create_app()


@app.shell_context_processor
def make_shell_context():
    return {'db': db, 'User': User}

再次运行 flask shell 命令:

(venv) D:\python-code\flask-vuejs-madblog\back-end>flask shell
Python 3.5.2 (v3.5.2:4def2a2901a5, Jun 25 2016, 22:18:55) [MSC v.1900 64 bit (AMD64)] on win32
App: app [production]
Instance: D:\python-code\flask-vuejs-madblog\back-end\instance
>>> app
<Flask 'app'>
>>> db
<SQLAlchemy engine=sqlite:///D:\python-code\flask-vuejs-madblog\back-end\app.db>
>>> User
<class 'app.models.User'>
>>> u = User(username='tom', email='tom@163.com')
>>> u.set_password('123456')
>>> u.check_password('123456')
True
>>> u.check_password('654321')
False
>>>

3. RESTful API设计

经过思考,我们的 用户资源 将暂时提供以下几个API:

HTTP方法 资源URL 说明
GET /api/users 返回所有用户的集合
POST /api/users 注册一个新用户
GET /api/users/<id> 返回一个用户
PUT /api/users/<id> 修改一个用户
DELETE /api/users/<id> 删除一个用户

创建 app/api/users.py

from app.api import bp


@bp.route('/users', methods=['POST'])
def create_user():
    '''注册一个新用户'''
    pass


@bp.route('/users', methods=['GET'])
def get_users():
    '''返回所有用户的集合'''
    pass


@bp.route('/users/<int:id>', methods=['GET'])
def get_user(id):
    '''返回一个用户'''
    pass


@bp.route('/users/<int:id>', methods=['PUT'])
def update_user(id):
    '''修改一个用户'''
    pass


@bp.route('/users/<int:id>', methods=['DELETE'])
def delete_user(id):
    '''删除一个用户'''
    pass

修改 app/api/__init__.py,在末尾添加:

from app.api import ping, users

3.1 用户对象转换成 JSON

后端 Flask 使用的都是 User 实例对象,而返回响应给前端时,需要传递 JSON 对象。修改 app/models.py,给 User 数据模型添加 to_dict 方法:

from flask import url_for
...

class User(db.Model):
    ...
    def to_dict(self, include_email=False):
        data = {
            'id': self.id,
            'username': self.username,
            '_links': {
                'self': url_for('api.get_user', id=self.id)
            }
        }
        if include_email:
            data['email'] = self.email
        return data

只有当用户请求自己的数据时才包含 email,使用 include_email 标志来确定该字段是否包含在字典中。后续调用该方法返回字典,再用 flask.jsonify 将字典转换成 JSON 响应

3.2 用户集合转换成 JSON

API 中有 POST /users 需要返回用户集合,所以还需要添加 to_collection_dict 方法。考虑到后续会创建 Post 等数据模型,所以在 app/models.py 中设计一个通用类 PaginatedAPIMixin

class PaginatedAPIMixin(object):
    @staticmethod
    def to_collection_dict(query, page, per_page, endpoint, **kwargs):
        resources = query.paginate(page, per_page, False)
        data = {
            'items': [item.to_dict() for item in resources.items],
            '_meta': {
                'page': page,
                'per_page': per_page,
                'total_pages': resources.pages,
                'total_items': resources.total
            },
            '_links': {
                'self': url_for(endpoint, page=page, per_page=per_page,
                                **kwargs),
                'next': url_for(endpoint, page=page + 1, per_page=per_page,
                                **kwargs) if resources.has_next else None,
                'prev': url_for(endpoint, page=page - 1, per_page=per_page,
                                **kwargs) if resources.has_prev else None
            }
        }
        return data

然后,User 继承这个类:

class User(PaginatedAPIMixin, db.Model):
    ...

3.3 JSON 转换成用户对象

前端发送过来 JSON 对象,需要转换成 User 对象

def from_dict(self, data, new_user=False):
    for field in ['username', 'email']:
        if field in data:
            setattr(self, field, data[field])
    if new_user and 'password' in data:
        self.set_password(data['password'])

说明: 不需要客户端发送用户列表到服务器,所以没有 JSON 转换成用户集合这个需求

3.4 错误处理

创建 app/api/errors.py

from flask import jsonify
from werkzeug.http import HTTP_STATUS_CODES


def error_response(status_code, message=None):
    payload = {'error': HTTP_STATUS_CODES.get(status_code, 'Unknown error')}
    if message:
        payload['message'] = message
    response = jsonify(payload)
    response.status_code = status_code
    return response


def bad_request(message):
    '''最常用的错误 400:错误的请求'''
    return error_response(400, message)

3.5 注册新用户

import re
from flask import request, jsonify, url_for
from app import db
from app.api import bp
from app.api.auth import token_auth
from app.api.errors import bad_request
from app.models import User


@bp.route('/users', methods=['POST'])
def create_user():
    '''注册一个新用户'''
    data = request.get_json()
    if not data:
        return bad_request('You must post JSON data.')

    message = {}
    if 'username' not in data or not data.get('username', None):
        message['username'] = 'Please provide a valid username.'
    pattern = '^(([^<>()\[\]\\.,;:\s@"]+(\.[^<>()\[\]\\.,;:\s@"]+)*)|(".+"))@((\[[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\])|(([a-zA-Z\-0-9]+\.)+[a-zA-Z]{2,}))$'
    if 'email' not in data or not re.match(pattern, data.get('email', None)):
        message['email'] = 'Please provide a valid email address.'
    if 'password' not in data or not data.get('password', None):
        message['password'] = 'Please provide a valid password.'

    if User.query.filter_by(username=data.get('username', None)).first():
        message['username'] = 'Please use a different username.'
    if User.query.filter_by(email=data.get('email', None)).first():
        message['email'] = 'Please use a different email address.'
    if message:
        return bad_request(message)

    user = User()
    user.from_dict(data, new_user=True)
    db.session.add(user)
    db.session.commit()
    response = jsonify(user.to_dict())
    response.status_code = 201
    # HTTP协议要求201响应包含一个值为新资源URL的Location头部
    response.headers['Location'] = url_for('api.get_user', id=user.id)
    return response

使用 HTTPiePostman 测试 API 接口

(venv) D:\python-code\flask-vuejs-madblog\back-end>pip install --upgrade httpie

create user api

3.6 检索单个用户

@bp.route('/users/<int:id>', methods=['GET'])
def get_user(id):
    '''返回一个用户'''
    return jsonify(User.query.get_or_404(id).to_dict())

get one user api

如果查询的用户 id 不存在,返回的是 404 错误页面:

get one user api error 01

我们要返回 JSON 错误信息,修改 app/api/errors.py

from app import db
from app.api import bp
...

@bp.app_errorhandler(404)
def not_found_error(error):
    return error_response(404)


@bp.app_errorhandler(500)
def internal_error(error):
    db.session.rollback()
    return error_response(500)

get one user api error 02

3.7 检索用户集合

@bp.route('/users', methods=['GET'])
def get_users():
    '''返回用户集合,分页'''
    page = request.args.get('page', 1, type=int)
    per_page = min(request.args.get('per_page', 10, type=int), 100)
    data = User.to_collection_dict(User.query, page, per_page, 'api.get_users')
    return jsonify(data)

先插入几条测试数据:

get multi users 01

注意: 三个测试用户的密码都是 123 ,但 hash 值不同

测试:

get multi users 02

3.8 修改用户

@bp.route('/users/<int:id>', methods=['PUT'])
def update_user(id):
    '''修改一个用户'''
    user = User.query.get_or_404(id)
    data = request.get_json()
    if not data:
        return bad_request('You must post JSON data.')

    message = {}
    if 'username' in data and not data.get('username', None):
        message['username'] = 'Please provide a valid username.'

    pattern = '^(([^<>()\[\]\\.,;:\s@"]+(\.[^<>()\[\]\\.,;:\s@"]+)*)|(".+"))@((\[[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\])|(([a-zA-Z\-0-9]+\.)+[a-zA-Z]{2,}))$'
    if 'email' in data and not re.match(pattern, data.get('email', None)):
        message['email'] = 'Please provide a valid email address.'

    if 'username' in data and data['username'] != user.username and \
            User.query.filter_by(username=data['username']).first():
        message['username'] = 'Please use a different username.'
    if 'email' in data and data['email'] != user.email and \
            User.query.filter_by(email=data['email']).first():
        message['email'] = 'Please use a different email address.'

    if message:
        return bad_request(message)

    user.from_dict(data, new_user=False)
    db.session.commit()
    return jsonify(user.to_dict())

modify one user

4. API 认证

为了简化使用 token 认证时客户端和服务器之间的交互,可以使用 Flask-HTTPAuth 插件

(venv) D:\python-code\flask-vuejs-madblog\back-end>pip install flask-httpauth
(venv) D:\python-code\flask-vuejs-madblog\back-end>pip freeze > requirements.txt

当客户端想要开始与 API 交互时,它需要使用用户名和密码进行 Basic Auth 验证,然后获得一个临时 token。只要 token 有效,客户端就可以发送附带 token 的 API 请求以通过认证。一旦 token 到期,需要申请新的 token

4.1 User 数据模型添加 token

修改 app/models.py

import base64
from datetime import datetime, timedelta
import os
...

class User(PaginatedAPIMixin, db.Model):
    ...
    token = db.Column(db.String(32), index=True, unique=True)
    token_expiration = db.Column(db.DateTime)
    ...

    def get_token(self, expires_in=3600):
        now = datetime.utcnow()
        if self.token and self.token_expiration > now + timedelta(seconds=60):
            return self.token
        self.token = base64.b64encode(os.urandom(24)).decode('utf-8')
        self.token_expiration = now + timedelta(seconds=expires_in)
        db.session.add(self)
        return self.token

    def revoke_token(self):
        self.token_expiration = datetime.utcnow() - timedelta(seconds=1)

    @staticmethod
    def check_token(token):
        user = User.query.filter_by(token=token).first()
        if user is None or user.token_expiration < datetime.utcnow():
            return None
        return user

创建数据库迁移脚本并应用:

(venv) D:\python-code\flask-vuejs-madblog\back-end>flask db migrate -m "user add tokens"
(venv) D:\python-code\flask-vuejs-madblog\back-end>flask db upgrade

4.2 HTTP Basic Authentication

创建 app/api/auth.py

from flask import g
from flask_httpauth import HTTPBasicAuth
from app.models import User
from app.api.errors import error_response

basic_auth = HTTPBasicAuth()


@basic_auth.verify_password
def verify_password(username, password):
    '''用于检查用户提供的用户名和密码'''
    user = User.query.filter_by(username=username).first()
    if user is None:
        return False
    g.current_user = user
    return user.check_password(password)


@basic_auth.error_handler
def basic_auth_error():
    '''用于在认证失败的情况下返回错误响应'''
    return error_response(401)

4.3 客户端申请 Token

现在我已经实现了 Basic Auth 验证的支持,因此我可以添加一条 token 检索路由,以便客户端在需要 token 时调用

创建 app/api/tokens.py

from flask import jsonify, g
from app import db
from app.api import bp
from app.api.auth import basic_auth


@bp.route('/tokens', methods=['POST'])
@basic_auth.login_required
def get_token():
    token = g.current_user.get_token()
    db.session.commit()
    return jsonify({'token': token})

装饰器 @basic_auth.login_required 将指示 Flask-HTTPAuth 验证身份,当通过 Basic Auth 验证后,才使用用户模型的 get_token() 方法来生成 token,数据库提交在生成 token 后发出,以确保 token 及其到期时间被写回到数据库

修改 app/api/__init__.py,在末尾添加:

from app.api import ping, users, tokens

如果你尝试直接向 token API 路由发送 POST 请求,则会发生以下情况:

flask httpauth 01

如果你在 POST 请求附带上了 Basic Auth 需要的凭证:

flask httpauth 02

4.4 HTTP Token Authentication

用户通过 Basic Auth 拿到 token 后,之后的请求只要附带这个 token 就能够访问其它 API,修改 app/api/auth.py

from flask_httpauth import HTTPBasicAuth, HTTPTokenAuth
...

token_auth = HTTPTokenAuth()
...

@token_auth.verify_token
def verify_token(token):
    '''用于检查用户请求是否有token,并且token真实存在,还在有效期内'''
    g.current_user = User.check_token(token) if token else None
    return g.current_user is not None


@token_auth.error_handler
def token_auth_error():
    '''用于在 Token Auth 认证失败的情况下返回错误响应'''
    return error_response(401)

4.5 使用 Token 机制保护 API 路由

create_user() 之外的所有 API 视图函数需要添加 @token_auth.login_required 装饰器,显而易见,这个函数不能使用 token 认证,因为用户都不存在时,更不会有 token 了

@bp.route('/users', methods=['GET'])
@token_auth.login_required
def get_users():
    ...

@bp.route('/users/<int:id>', methods=['GET'])
@token_auth.login_required
def get_user(id):
    ...

...

如果你直接对上面列出的受 token 保护的 endpoint 发起请求,则会得到一个 401 错误:

flask httpauth 03

为了成功访问,你需要添加 Authorization 头部,其值是请求 /api/tokens 获得的 token 的值:

flask httpauth 04

4.6 撤销 Token

修改 app/api/tokens.py

from app.api.auth import basic_auth, token_auth
...

@bp.route('/tokens', methods=['DELETE'])
@token_auth.login_required
def revoke_token():
    g.current_user.revoke_token()
    db.session.commit()
    return '', 204

客户端可以向 /api/tokens URL发送 DELETE 请求,以使 token 失效

5. 提交代码

最终代码结构如下:

$ git add .
$ git commit -m "3. Flask设计User用户相关API"
$ git checkout master
$ git merge dev
$ git branch -d dev

将本地 master 分支代码上传到 Github 代码仓库中的 master 分支:

$ git push -u origin master

打上标签 tag并上传:

$ git tag v0.3
$ git push origin v0.3

代码已上传到 https://github.com/wangy8961/flask-vuejs-madblog/tree/v0.3 ,欢迎star

分类: Vue.js
标签: API RESTful vuejs flask
未经允许不得转载: LIFE & SHARE - 王颜公子 » Flask Vue.js全栈开发|第3章:Flask设计User用户相关API

分享

作者

作者头像

Madman

如果博文内容有误或其它任何问题,欢迎留言评论,我会尽快回复; 或者通过QQ、微信等联系我

0 条评论

暂时还没有评论.

发表评论前请先登录

专题