本章基于 token
认证,添加 创建用户、获取单个/所有用户、修改用户、删除用户 等 API
接口,测试工具 HTTPie/Postman
。
1. 拉取最新代码
# 查看远程地址
$ git remote -v
origin https://gitee.com/hubery_jun/flask-vuejs-madblog (fetch)
origin https://gitee.com/hubery_jun/flask-vuejs-madblog (push)
# 类似于 git pull,也是用于拉取最新代码
$ git fetch
# 或拉取指定的远程主机上的分支,如 origin 上的 master
$ git fetch origin master
git fetch 与 git pull 的区别
git fetch
:- 远端跟踪分支:可以更改远端跟踪分支
- 拉取:会将数据拉取到本地仓库,但是不会自动合并或修改当前的工作
commitID
:本地库中master
的commitID
不变,还是等于 1
git pull
:- 远端跟踪分支:无法对远端跟踪分支操作,必须先切回到本地分支然后创建一个新的
commit
提交 - 拉取:从远处获取最新版本,并合并到本地,会自动合并或修改当前的工作
commitID
:本地库中master
的commitID
发生改变,变成了 2
- 远端跟踪分支:无法对远端跟踪分支操作,必须先切回到本地分支然后创建一个新的
创建 dev 分支
git checkout -b dev
git branch
2. 用户模型设计
2.1 使用 ORM SQLAlchemy
两个插件:
flask-sqlalchemy
:ORM
相关Flask-Migrate
:用于迁移数据表结构
1、安装:
pip install flask-sqlalchemy flask-migrate
pip freeze > requirements.txt
2、配置 SQLite
数据库,修改 back-end/config.py
:
import os
from dotenv import load_dotenv
basedir = os.path.abspath(os.path.dirname(__file__))
load_dotenv(os.path.join(basedir, '.env'), encoding='utf-8')
class Config(object):
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
'sqlite:///' + os.path.join(basedir, 'app.db')
SQLALCHEMY_TRACK_MODIFICATIONS = False
注意:迁移成功后,会生成一个
back-end/app.db
数据库文件,可以使用Navicat
可视化工具打开!
3、初始化数据库,app/__init__.py
:
# 数据库相关
db = SQLAlchemy()
migrate = Migrate()
def create_app(config_class=Config):
app = Flask(__name__)
app.config.from_object(config_class)
# 跨域
CORS(app)
# 初始化数据库
db.init_app(app)
migrate.init_app(app, db)
# 注册蓝图 blueprint
from app.api import bp as api_bp
app.register_blueprint(api_bp, url_prefix="/api")
return app
from app import models
2.2 定义用户模型
1、创建 app/models.py
:
class User(db.Model):
"""用户对象"""
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(64), index=True, unique=True) # index 创建索引
email = db.Column(db.String(120), index=True, unique=True)
password_hash = db.Column(db.String(128)) # 密码加密(hash),不存明文
def __str__(self):
return '<User {}>'.format(self.username)
2、创建迁移存储库:
(flask-vuejs) F:\My Projects\flask-vuejs-madblog\back-end> flask db init
3、生成迁移脚本:
# -m 参数:添加记录
(flask-vuejs) F:\My Projects\flask-vuejs-madblog\back-end> flask db migrate -m "add users table"
2、将迁移脚本应用到数据库中:
# flask db upgrade 还可以回滚到上次的迁移,需要指定
(flask-vuejs) F:\My Projects\flask-vuejs-madblog\back-end> flask db upgrade
2.3 密码哈希
在数据表中,不能直接保存明文密码,这里我们将使用 werkzeug.security
库的 generate_password_hash
和 check_password_hash
来创建哈希密码和验证密码的 hash
是否一致。
更新 app/models.py
:
from werkzeug.security import generate_password_hash, check_password_hash
class User(PaginationAPIMixin, db.Model):
"""用户对象"""
...
def generate_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
可以与项目环境进行交互(会启动一个 Python
解释器包含应用的上下文),默认不支持 db
数据库模型的使用,需要额外配置。
1、修改 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():
"""配置flask shell 上下文"""
return {'db': db, 'User': User}
2、在终端进入 flask shell
:
(flask-vuejs) F:\My Projects\flask-vuejs-madblog\back-end>flask shell
Python 3.6.8 (tags/v3.6.8:3c6b436a57, Dec 24 2018, 00:16:47) [MSC v.1916 64 bit (AMD64)] on win32
App: app [production]
Instance: F:\My Projects\flask-vuejs-madblog\back-end\instance
>>> app
<Flask 'app'>
>>> db
<SQLAlchemy engine=sqlite:///F:\My Projects\flask-vuejs-madblog\back-end\app.db>
>>> User
<class 'app.models.User'>
>>> u = User(username='rose', email='rose@qq.com')
>>> u.generate_password('123456')
>>> u.check_password('123456')
True
注意:需要先进入项目虚拟环境!
3. 用户相关 API 设计
用户资源相关的 api
:
HTTP方法 | 资源URL | 说明 |
---|---|---|
GET | /api/users | 返回所有用户的集合 |
POST | /api/users | 注册一个新用户 |
GET | /api/users/ |
返回一个用户 |
PUT | /api/users/ |
修改一个用户 |
DELETE | /api/users/ |
删除一个用户 |
新建: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
记得要将 users
添加到 api/__init__.py
:
from app.api import ping, users
3.1 用户对象转换成 JSON
因为 API
接口返回给前端的数据为 json
数据,所以封装 User
模型为 json
形式,方便传递,app/models.py
新增:
class User(db.Model):
"""用户对象"""
...
def to_dict(self, include_email=False):
"""
封装 User 对象,传递给前端只能是 json 格式,不能是实例对象
:param include_email: 只有当用户请求自己数据时,才包含 email
:return:
"""
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
include_email
用来标记 email
字段是否在字典中,只有当用户请求自己的数据时,才包含。
3.2 用户集合转换为 JSON
当获取所有用户数据时也需要封装为 json
形式,另外还包含了分页信息,为了后续能够重复利用,将其设计为通用设计类,app/models.py
:
import base64
import os
from datetime import datetime, timedelta
from flask import url_for
from app import db
from werkzeug.security import generate_password_hash, check_password_hash
class PaginationAPIMixin:
@staticmethod
def to_collection_dict(query, page, per_page, endpoint, **kwargs):
# 分页查询,error_out 表示页数不是 int 或 超过总页数时,会报错,并返回 404,默认为 True
resources = query.paginate(page, per_page, error_out=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), # "/api/users?page=1&per_page=10"
'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(PaginationAPIMixin, db.Model):
"""用户对象"""
3.3 JSON 转换为用户对象
将前端传过来的 JSON
数据转换为 User
对象,app/models.py
:
class User(PaginationAPIMixin, db.Model):
"""用户对象"""
....
def from_dict(self, data, new_user=False):
"""
将前端发送过来的 json 对象转换为 User 对象
:param data:
:param new_user:
:return:
"""
for field in ['username', 'email']:
if field in data:
# 给实例对象添加属性字典
setattr(self, field, data[field])
if new_user and 'password' in data:
self.generate_password(data['password'])
3.4 错误处理
创建 app/api/errors.py
:
from flask import jsonify
from werkzeug.http import HTTP_STATUS_CODES
from app import db
from app.api import bp
def error_response(status_code, message=None):
payload = {'error': HTTP_STATUS_CODES.get(status_code, 'Unknow error')}
if message:
payload['message'] = message
response = jsonify(payload)
response.status_code = status_code
return response
def bad_request(message):
"""
异常请求,如:400
:param message:
:return:
"""
return error_response(400, message)
3.5 创建新用户
编辑 app/api/users.py
:
@bp.route('/users', methods=['POST'])
def create_user():
"""创建一个新用户"""
data = request.get_json()
if not data:
return bad_request("post 必须是 json 数据!")
message = {}
username = data.get('username', None)
email = data.get('email', None)
password = data.get('password', None)
# 判断是否为空
if 'username' not in data or not username:
message['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, email):
message['email'] = "请提供一个有效的邮箱地址!"
if 'password' not in data or not password:
message['password'] = "请提供一个有效的密码!"
# 检查数据库中是否有该用户
if User.query.filter(or_(User.username == username, User.email == email)).first():
message['username'] = "用户名或邮箱已存在!"
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
response.headers['Location'] = url_for('api.get_user', id=user.id) # /api/users/1
return response
使用 HTTPie
模块来测试 API
接口:
pip install --upgrade httpie
pip freeze > requirements.txt
测试结果:
(flask-vuejs) F:\Envs\flask-vuejs\Scripts>http POST http://127.0.0.1:5000/api/users username=john password=123456 email=john@qq.com
HTTP/1.0 201 CREATED
Access-Control-Allow-Origin: *
Content-Length: 60
Content-Type: application/json
Date: Mon, 31 Aug 2020 02:42:49 GMT
Location: http://127.0.0.1:5000/api/users/3
Server: Werkzeug/1.0.1 Python/3.6.8
{
"_links": {
"self": "/api/users/3"
},
"id": 3,
"username": "john"
}
3.6 查询单个用户
编辑 app/api/users.py
:
@bp.route('/users/<int:id>', methods=['GET'])
def get_user(id):
"""返回一个用户"""
return jsonify(User.query.get_or_404(id).to_dict())
测试结果:
(flask-vuejs) F:\Envs\flask-vuejs\Scripts>http GET http://127.0.0.1:5000/api/users/3
HTTP/1.0 200 OK
Access-Control-Allow-Origin: *
Content-Length: 60
Content-Type: application/json
Date: Mon, 31 Aug 2020 02:43:34 GMT
Server: Werkzeug/1.0.1 Python/3.6.8
{
"_links": {
"self": "/api/users/3"
},
"id": 3,
"username": "john"
}
可以看到返回的就是 to_dict()
封装的数据。
构造查询不存在时返回的数据
当查询不存在的用户,也返回一个 JSON
数据,修改 app/api/errors.py
,新增:
@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)
测试结果:
# 测试不存在的用户
(flask-vuejs) F:\Envs\flask-vuejs\Scripts>http GET http://127.0.0.1:5000/api/users/36
HTTP/1.0 404 NOT FOUND
Access-Control-Allow-Origin: *
Content-Length: 22
Content-Type: application/json
Date: Mon, 31 Aug 2020 02:43:53 GMT
Server: Werkzeug/1.0.1 Python/3.6.8
{
"error": "Not Found"
}
3.7 查询所有用户
编辑 app/api/users.py
,新增:
@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)
page
为当前页码数,per_page
为每页要显示的条数。
测试结果:
(flask-vuejs) F:\Envs\flask-vuejs\Scripts>http GET http://127.0.0.1:5000/api/users
HTTP/1.0 200 OK
Access-Control-Allow-Origin: *
Content-Length: 331
Content-Type: application/json
Date: Mon, 31 Aug 2020 02:44:57 GMT
Server: Werkzeug/1.0.1 Python/3.6.8
{
"_links": {
"next": null,
"prev": null,
"self": "/api/users?page=1&per_page=10"
},
"_meta": {
"page": 1,
"per_page": 10,
"total_items": 3,
"total_pages": 1
},
"items": [
{
"_links": {
"self": "/api/users/1"
},
"id": 1,
"username": "rose"
},
{
"_links": {
"self": "/api/users/2"
},
"id": 2,
"username": "lila"
},
{
"_links": {
"self": "/api/users/3"
},
"id": 3,
"username": "john"
}
]
}
3.8 修改用户
编辑 app/api/users.py
,新增:
@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("post 必须是 json 数据!")
message = {}
username = data.get('username', None)
email = data.get('email', None)
# 判断是否为空
if 'username' in data and not username:
message['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, email):
message['email'] = "请提供一个有效的邮箱地址!"
if 'username' in data and data['username'] != user.username and \
User.query.filter_by(username=data['username']).first():
message['username'] = '请使用一个不同的用户名!'
if 'email' in data and data['email'] != user.email and \
User.query.filter_by(email=data['email']).first():
message['email'] = '请使用一个不同的邮箱!'
if message:
return bad_request(message)
user.from_dict(data, new_user=False)
db.session.commit()
return jsonify(user.to_dict())
测试结果:
# 输入要修改的用户 ID 和要修改的字段
(flask-vuejs) F:\Envs\flask-vuejs\Scripts>http PUT http://127.0.0.1:5000/api/users/3 email=john@outlook.com
HTTP/1.0 200 OK
Access-Control-Allow-Origin: *
Content-Length: 60
Content-Type: application/json
Date: Mon, 31 Aug 2020 02:49:37 GMT
Server: Werkzeug/1.0.1 Python/3.6.8
{
"_links": {
"self": "/api/users/3"
},
"id": 3,
"username": "john"
}
4. API 认证
所谓 API
认证,即只有得到认证过的请求,才能访问特定的 API
,比如(登录认证、token
认证等),这里采用的是 Flask-HTTPAuth
模块。
它需要使用用户名和密码进行 Basic Auth
验证,然后获得一个临时 token
。只要 token
有效,客户端就可以发送附带 token
的 API 请求以通过认证。一旦 token
到期,需要申请新的 token
。
安装:
pip install flask-httpauth
pip freeze > requirements.txt
4.1 User 用户模型添加 token
编辑 app/models.py
:
import base64
import os
from datetime import datetime, timedelta
class User(PaginationAPIMixin, db.Model):
"""用户对象"""
....
# token 验证 API(需要登录才能请求)
token = db.Column(db.String(32), index=True, unique=True)
token_expiration = db.Column(db.DateTime) # token 过期时间
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') # 生成 token
self.token_expiration = now + timedelta(seconds=expires_in)
db.session.add(self)
return self.token
def revoke_token(self):
"""撤销 token,当前 utc 时间减去 1 秒"""
self.token_expiration = datetime.utcnow() - timedelta(seconds=1)
@staticmethod
def check_token(token):
"""检查 token"""
user = User.query.filter_by(token=token).first()
# 若没有 token 或者 token 已过期,返回 None,不准请求
if user is None or user.token_expiration < datetime.utcnow():
return None
return user
因为新增了字段,所以需要迁移生成新的数据表:
flask db migrate -m "user add tokens"
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
路由,创建 app/api/tokens.py
:
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
,直接请求,会提示需要登录:
(flask-vuejs) F:\Envs\flask-vuejs\Scripts>http POST http://127.0.0.1:5000/api/tokens
HTTP/1.0 401 UNAUTHORIZED
Access-Control-Allow-Origin: *
Content-Length: 25
Content-Type: application/json
Date: Mon, 31 Aug 2020 02:50:32 GMT
Server: Werkzeug/1.0.1 Python/3.6.8
WWW-Authenticate: Basic realm="Authentication Required"
{
"error": "Unauthorized"
}
需要带上用户登录信息:
(flask-vuejs) F:\Envs\flask-vuejs\Scripts>http --auth john:123456 POST http://127.0.0.1:5000/api/tokens
HTTP/1.0 200 OK
Access-Control-Allow-Origin: *
Content-Length: 45
Content-Type: application/json
Date: Mon, 31 Aug 2020 02:51:15 GMT
Server: Werkzeug/1.0.1 Python/3.6.8
{
"token": "G4d8FwoEdOODyhBBe8nz30vCe0X+YUAI"
}
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 路由
除了创建用户不用保护以后,其他路由都需要 Token
保护,app/api/users.py
:
@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_auth.login_required
装饰器即可。
测试
为携带 token
的请求,会得到一个 401
的错误:
(flask-vuejs) F:\Envs\flask-vuejs\Scripts>http GET http://127.0.0.1:5000/api/users/3
HTTP/1.0 401 UNAUTHORIZED
Access-Control-Allow-Origin: *
Content-Length: 25
Content-Type: application/json
Date: Mon, 31 Aug 2020 02:54:38 GMT
Server: Werkzeug/1.0.1 Python/3.6.8
WWW-Authenticate: Bearer realm="Authentication Required"
{
"error": "Unauthorized"
}
携带 token
的请求,返回 200:
# 需要添加 Authorization 头部
(flask-vuejs) F:\Envs\flask-vuejs\Scripts>http GET http://127.0.0.1:5000/api/users/3 "Authorization:Bearer G4d8FwoEdOODyhBBe8nz30vCe0X+YUAI"
HTTP/1.0 200 OK
Access-Control-Allow-Origin: *
Content-Length: 60
Content-Type: application/json
Date: Mon, 31 Aug 2020 02:55:20 GMT
Server: Werkzeug/1.0.1 Python/3.6.8
{
"_links": {
"self": "/api/users/3"
},
"id": 3,
"username": "john"
}
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
测试:
# 删除 token
(flask-vuejs) F:\Envs\flask-vuejs\Scripts>http DELETE http://127.0.0.1:5000/api/tokens "Authorization:Bearer G4d8FwoEdOODyhBBe8nz30vCe0X+YUAI"
HTTP/1.0 204 NO CONTENT
Access-Control-Allow-Origin: *
Content-Type: text/html; charset=utf-8
Date: Mon, 31 Aug 2020 03:02:08 GMT
Server: Werkzeug/1.0.1 Python/3.6.8
# 再使用这条 token 进行请求,发现请求失败
(flask-vuejs) F:\Envs\flask-vuejs\Scripts>http GET http://127.0.0.1:5000/api/users/3 "Authorization:Bearer G4d8FwoEdOODyhBBe8nz30vCe0X+YUAI"
HTTP/1.0 401 UNAUTHORIZED
Access-Control-Allow-Origin: *
Content-Length: 25
Content-Type: application/json
Date: Mon, 31 Aug 2020 03:02:18 GMT
Server: Werkzeug/1.0.1 Python/3.6.8
WWW-Authenticate: Bearer realm="Authentication Required"
{
"error": "Unauthorized"
}
5. 提交代码
项目结构:
back-end/
├─app
│ ├─api
│ │ └─__init__.py
│ │ └─auth.py
│ │ └─errors.py
│ │ └─ping.py
│ │ └─tokens.py
│ │ └─users.py
│ └─__init__.py__
│ └─models.py__
├─migrations
└─.env
└─.gitignore
└─app.db
└─config.py
└─madblog.py
└─requirements.txt
合并分支并推送到远端
$ git add .
$ git commit -m "3. Flask设计User用户相关API"
$ git checkout master
$ git merge dev
$ git branch -d dev
$ git push -u origin master
打标签
$ git tag v0.3
hj@DESKTOP-JUS39UG MINGW32 /f/My Projects/flask-vuejs-madblog (master)
$ git push origin v0.3
Total 0 (delta 0), reused 0 (delta 0), pack-reused 0
remote: Powered by GITEE.COM [GNK-5.0]
To https://gitee.com/hubery_jun/flask-vuejs-madblog
* [new tag] v0.3 -> v0.3