测试之家

通用技术 测试开发——Flask 入门教程系列：04-加入注册登录

弗多 · 2018年02月28日 · 最后由 mark 回复于 2018年10月05日 · 2735 次阅读

传送门：系列教程前言及整理贴 测试开发——Flask 入门教程系列 (整理及前言)

使用 Flask-Login 注册登录

上一章我们为 TODO APP 引入了数据库，若我们开启服务至外网，任何人都可以操作，这就引入了用户登录概念。目前几乎所有网站、APP 都需要注册登录。所以，如何加入注册登录是我们必须要了解及掌握的。

本章将讲解如何使用 Flask 的又一扩展 Flask-Login 来进行访问控制。可查阅：Flask-Login 官方文档 。

用户模型

我们先不用急于学习 Flask-Login，而是回顾上一章的内容，创建下用户模型。

养成好习惯，先设想下用户的数据，从而设计用户数据结构：

"user": {
  "user_id": 1,
  "name": "Mike",
  "email": "Mike@gmail.com", 
  "pwd": "123456ab",
  "createtime": 2018-02-17 17:02:19.15
}

！注意：这边 pwd 未加密存储，实际生产中加密是必须的，至于如何加密这边就不深入了。

这边先写两个简单的 API 用以 获取用户信息以及 新增用户（后续会改写这两个 API），API 设计如下：

POST /user 新增用户

请求传参：

​ Headers Content-Type: application/json

​ 请求体为 Json 格式，name、pwd 为必传项，email 为选传项；若已存在同名的用户，返回错误

​ 示例：

{
    "name": "Mike",
    "email": "Mike@gmail.com",
    "pwd": "123456ab"
}

返回示例：

{
    "status": 0,
    "user_id": 1
}

若传参错误，则返回：

{
    "err": "Request not Json or miss name/pwd."
}

若 name 已存在，则返回：

{
    "err": "Name is already existed."
}

若数据存储错误，则返回：

{
    "err": "Save Error. Please check your input length: pwd>6, name<100, email<200."
}

GET /user/user_id 获取指定用户的信息

请求传参：URL 中需带入 user_id

返回示例：

{
    "status": 0,
    "user": {
        "email": "Mike@gmail.com",
        "name": "Mike",
        "user_id": 1
    }
}

若对应 id 的 user 不存在，则返回：

{
    "err": "Not found."
}

若已掌握上一章内容，代码并不难写（可先不要看，而是试着自己写下，就当回顾上节学习内容了）：

#!/usr/bin/env python3

from flask import Flask, jsonify, request
from flask_mongoengine import MongoEngine

from datetime import datetime


app = Flask(__name__)
app.config['MONGODB_SETTINGS'] = {
    'db': 'todo',
    'host': 'localhost',
    'port': 27017
}

db = MongoEngine()
db.init_app(app)


class User(db.Document):
    user_id = db.IntField(required=True)
    name = db.StringField(required=True, max_length=100)
    email = db.StringField(max_length=200)
    pwd = db.StringField(requied=True, min_length=6)
    createtime = db.DateTimeField(required=True)

    def to_json(self):
        return {
            "user_id": self.user_id,
            "name": self.name,
            "email": self.email
        }


@app.route('/user', methods=['POST'])
def postUser():
    if not request.json or not 'name' in request.json or not 'pwd' in request.json:
        return jsonify({'err': 'Request not Json or miss name or pwd'})
    elif User.objects(name=request.json['name']).first():
        return jsonify({'err': 'Name is already existed.'})
    else:
        try:
            user = User(
                user_id=User.objects().count() + 1,
                name=request.json['name'],
                email=request.json['email'] if 'email' in request.json else "",
                pwd=request.json['pwd'],
                createtime=datetime.now()
            )
            user.save()
        except Exception as e:
            return jsonify({'err': 'Save Error. Please check your input length: pwd>6, name<100, email<200.'})
    return jsonify({'status': 0, 'user_id': user['user_id']})


@app.route('/user/<int:user_id>', methods=['GET'])
def getUser(user_id):
    user = User.objects(user_id=user_id).first()
    if not user:
        return jsonify({'err': 'Not found.'})
    else:
        return jsonify({'status': 0, 'user': user.to_json()})


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

注册、登录功能及 API 设计

简单的注册登录功能应该都不陌生，我们设计注册、登录及用户信息等基础功能如下：

  • 用户注册：用户名（不可重名、必填）、邮箱（非必填）、密码（大于 6 位）；注册后保存用户数据
  • 用户登录：已有用户名及密码匹配后登录成功，不然提示失败；登录成功后保留登录态
  • 用户登出：当前登录的用户登录态失效
  • 用户信息：显示当前登录用户的用户名及邮箱，若用户未登录则报错提示
  • 更改邮箱：仅登录态可更改，更改用户邮箱为新输入的邮箱
  • 更改密码：仅登录态可更改，要求输入原密码及新密码，若成功，更改密码为新密码

实际生产中，注册、登录的涉及内容很多，且大多涉及账户安全，如验证码、密码加密、邮箱找回密码等，这边就不深究了。（有兴趣的可以 Google 自学，目前已有不少成熟的解决方案）

总之，我们设计 API 如下：

POST /register 用户注册

请求传参：

​ Headers Content-Type: application/json

​ 请求体为 Json 格式，name、pwd 为必传项，email 为选传项

​ 示例：

{
  "name": "Mike",
  "email": "Mike@gmail.com",
  "pwd": "123456ab"
}

返回示例：

{
  "status": 0,
  "user_id": 1,
  "msg": "Register success."
}

若传参缺省错误，则返回：

{
    "err": "Request not Json or miss name/pwd."
}

若 name 已存在，则返回：

{
    "err": "Name is already existed."
}

若数据存储错误，则返回：

用户输入数据长度不满足数据模型要求也会返回该错，数据长度的限制前后端均都要有。

{
    "err": "Register error."
}

POST /login 用户登录

请求传参：

​ Headers Content-Type: application/json

​ 请求体为 Json 格式，name、pwd 为必传项

​ 示例：

{
  "name": "Mike",
  "pwd": "123456ab"
}

返回示例：

{
  "status": 0,
  "user_id": 1,
  "msg": "Login success."
}

若登录失败，则返回：

{
  "err": "Login fail."
}

POST /logout 用户登出

返回示例：

{
  "status": 0,
  "msg": "Logout success."
}

GET /user 获取当前登录的用户信息

返回示例：

{
    "status": 0,
    "user": {
        "email": "Mike@gmail.com",
        "name": "Mike",
        "user_id": 1
    }
}

若用户未登录，则返回：

{
    "err": "Not login."
}

PUT /user/email 当前登录的用户更改邮箱

请求传参：email 必传

{
    "email": "Mike@163.com"
}

返回示例：

{
    "msg": "Email has been modified.",
    "status": 0,
    "user": {
        "email": "Mike@163.com",
        "name": "Mike",
        "user_id": 1
    }
}

PUT /user/pwd 当前登录的用户更改密码

请求传参：current_pwd - 原密码；new_pwd - 新密码；

{
    "current_pwd": "123456ab",
    "new_pwd": "123456ba"
}

返回示例：

{
  "status": 0,
  "user_id": 1,
  "msg": "PWD has been modified."
}

若用户输入的原密码错误，则返回：

{
    "err": "current_pwd is not right."
}

引入 Flask-Login

初始化 Flask-Login 及设置

如同使用 Flask-MongoEngine 一样，Flask-Login 也需要导入以及和 app 服务器进行初始化绑定关联，如下：

from flask_login import LoginManager
login_manager = LoginManager()
login_manager.init_app(app)

这样就将 Flask-Login 和服务器绑定起来了。但是，Flask-Login 怎么才知道登录的 URL 的是哪个？怎么验证我们的账号密码？怎么才能知道登陆的用户是谁？这些都是关键的问题，别急，慢慢一一道来。

注册用户

注册 API 逻辑不涉及登录态，但我们需要改写下之前 postUser() 的部分名称及文字：

@app.route('/register', methods=['POST'])
def registerUser():
    if not request.json or not 'name' in request.json or not 'pwd' in request.json:
        return jsonify({'err': 'Request not Json or miss name/pwd'})
    elif User.objects(name=request.json['name']).first():
        return jsonify({'err': 'Name is already existed.'})
    else:
        user_id = User.objects().count() + 1
        user = User(
            uuid=shortuuid.uuid(),
            user_id=user_id,
            name=request.json['name'],
            email=request.json['email'] if 'email' in request.json else "",
            pwd=request.json['pwd'],
            createtime=datetime.now()
        )
        try:
            user.save()
        except Exception:
            return jsonify({'err': 'Register error.'})
    return jsonify({'status': 0, 'user_id': user['user_id'], 'msg': 'Register success.'})

用户登录

问题还是比较多的，例如登陆的 URL 是什么？登录 API 后如何成功后保留用户登录态？先看新增的代码吧：

...

from flask.ext.login import LoginManager, login_user

...

login_manager = LoginManager()
login_manager.init_app(app)
app.secret_key = 'secret_key'

class User(db.Document):

    ...

    def is_authenticated(self):
        return True

    def is_active(self):
        return True

    def is_anonymous(self):
        return False

    def get_id(self):
        return str(self.user_id)

 ...

@user4oader
def load_user(user_id):
    return User.objects(user_id=user_id).first()

@app.route('/login', methods=['POST'])
def login():
    if not request.json or not 'name' in request.json or not 'pwd' in request.json:
        return jsonify({'err': 'Request not Json or miss name/pwd'})
    else:
        user = User.objects(
            name=request.json['name'], pwd=request.json['pwd']).first()
    if user:
        login_user(user)
        return jsonify({'status': 0, 'user_id': user.get_id(), 'msg': 'Login success.'})
    else:
        return jsonify({'err': 'Login fail.'})

就以上代码加以讲解：

  1. from flask_login import LoginManager, login_user ，LoginManager 处理 app 初始化绑定；

login_user 函数：将当前用户的状态设置成已登录（flask-login.login_user）

  1. app.secret_key = 'itsasecrectkey' 加密 session 用

  2. class User(db.Document):
   ...
   def is_authenticated(self):
       return True

   def is_active(self):
       return True

   def is_anonymous(self):
       return False

   def get_id(self):
       return str(self.user_id)

这边我们完善了 User Class， 简单介绍如下：（ flask-login user class ）

  • is_authenticated：当前用户是否被授权，因为我们登陆了就可以操作，所以默认都是被授权的
  • is_anonymous: 用于判断当前用户是否是匿名用户，登录后的用户自然不是匿名用户
  • is_active： 用于判断当前用户是否已经激活，已经激活的用户才能登陆
  • get_id： 获取该用户的唯一标示
  1. python @login_manager.user_loader def load_user(user_id): return User.objects(user_id=user_id).first()

需要告诉 Flask-Login 如何通过一个 id 获取用户，通过指定 user_loader，就可以查询到当前的登陆用户。（flask_login.LoginManager.user_loader）

实际用户每次登录登出都可以记录下来便于用户行为分析，像是登录设备、时间等，这边就不多加详述了。

注册成功是否保留登录态？视具体情况来定，若需要，可以在注册用户的代码里加上 login_user(user) 即可

用户登出

完成了用户登录，是不是觉得 Flask-Login 很强大？用户登出就更不难理解了，仅需一个函数：

​ logout_user() 函数：当前登录用户退出登录状态 (flask_login.logout_user)

使用该函数，该 API 的代码就完成了：

from flask.ext.login import LoginManager, login_user, logout_user
...

@app.route('/logout', methods=['POST'])
def logout():
    logout_user()
    return jsonify({'status': 0, 'msg':'Logout success.'})

用户信息

用户登录态保留后，并不需传递 user_id 参数，故改写之前的 getUser()。

这边再引入一个重要概念，即 current_user 该变量表示当前登录的用户：

​ 若已登陆，那么它就是我们设置的 Model User 的对象， is_authenticated 为 True；

​ 若未登陆，那么 is_authenticated 就为 False。 (flask_login.current_user)

代码修改如下：

from flask_login import LoginManager, login_user, logout_user, current_user
...

@app.route('/user', methods=['GET'])
def getUser():
    if current_user.is_authenticated:
        return jsonify({'status': 0, 'user': current_user.to_json()})
    else:
        return jsonify({'err': 'Not login.'})

写好了该 API，我们可以进一步验证 登录/登出 API 的功能是否正确了~（之前写的没测试总感觉没谱啊哈哈）

更改用户

更改用户的密码、邮箱，此类型操作要求仅登录态才可见（未登录会报 401 错误），实际上很多 API 都有此要求。

Flask-Login 处理起来相当容易，只需要加 login_required 的装饰器即可（flask_login.login_required）

同样的，我们引入该函数并写更改用户邮箱及密码的 API 如下：

from flask_login import LoginManager, login_user, logout_user, current_user, login_required

...

@app.route('/user/pwd', methods=['PUT'])
@login_required
def putUserPWD():
    if not request.json or not 'current_pwd' in request.json or not 'new_pwd' in request.json:
        return jsonify({'err': 'Request not Json or miss current_pwd/new_pwd'})
    else:
        current_pwd = current_user.pwd
    if not request.json['current_pwd'] == current_pwd:
        return jsonify({'err': 'current_pwd is not right.'})
    else:
        current_user.pwd = request.json['new_pwd']
        try:
            current_user.save()
        except Exception:
            return jsonify({'err': 'Modify PWD error.'})
        return jsonify({'status': 0, 'msg': 'PWD has been modified.', 'user_id': current_user.user_id})

完整代码

完整的代码如下，基本实现了用户的注册、登录、登出，修改邮箱及密码。

#!/usr/bin/env python3

from flask import Flask, jsonify, request
from flask_mongoengine import MongoEngine
from flask_login import LoginManager, login_user, logout_user, current_user, login_required

from datetime import datetime

app = Flask(__name__)
app.config['MONGODB_SETTINGS'] = {
    'db': 'user',
    'host': 'localhost',
    'port': 27017
}

db = MongoEngine()
db.init_app(app)

login_manager = LoginManager()
login_manager.init_app(app)
app.secret_key = 'secret_key'


class User(db.Document):
    user_id = db.IntField(required=True)
    name = db.StringField(required=True, max_length=100)
    email = db.StringField(max_length=200)
    pwd = db.StringField(requied=True, min_length=6)
    createtime = db.DateTimeField(required=True)

    def to_json(self):
        return {
            "user_id": self.user_id,
            "name": self.name,
            "email": self.email
        }

    def is_authenticated(self):
        return True

    def is_active(self):
        return True

    def is_anonymous(self):
        return False

    def get_id(self):
        return str(self.user_id)


@user11oader
def load_user(user_id):
    return User.objects(user_id=user_id).first()


@app.route('/register', methods=['POST'])
def registerUser():
    if not request.json or not 'name' in request.json or not 'pwd' in request.json:
        return jsonify({'err': 'Request not Json or miss name/pwd'})
    elif User.objects(name=request.json['name']).first():
        return jsonify({'err': 'Name is already existed.'})
    else:
        user = User(
            user_id=User.objects().count() + 1,
            name=request.json['name'],
            email=request.json['email'] if 'email' in request.json else "",
            pwd=request.json['pwd'],
            createtime=datetime.now()
        )
        try:
            user.save()
        except Exception:
            return jsonify({'err': 'Register error.'})
    return jsonify({'status': 0, 'user_id': user['user_id'], 'msg': 'Register success.'})


@app.route('/login', methods=['POST'])
def login():
    if not request.json or not 'name' in request.json or not 'pwd' in request.json:
        return jsonify({'err': 'Request not Json or miss name/pwd'})
    else:
        user = User.objects(
            name=request.json['name'], pwd=request.json['pwd']).first()
    if user:
        login_user(user)
        return jsonify({'status': 0, 'user_id': user.get_id(), 'msg': 'Login success.'})
    else:
        return jsonify({'err': 'Login fail.'})


@app.route('/logout', methods=['POST'])
def logout():
    logout_user()
    return jsonify({'status': 0, 'msg': 'Logout success.'})


@app.route('/user', methods=['GET'])
def getUser():
    if current_user.is_authenticated:
        return jsonify({'status': 0, 'user': current_user.to_json()})
    else:
        return jsonify({'err': 'Not login.'})


@app.route('/user/email', methods=['PUT'])
@login_required
def putUserEmail():
    if not request.json or not 'email' in request.json:
        return jsonify({'err': 'Request not Json or miss email'})
    else:
        current_user.email = request.json['email']
        try:
            current_user.save()
        except Exception:
            return jsonify({'err': 'Modify email error.'})
        return jsonify({'status': 0, 'msg': 'Email has been modified.', 'user': current_user.to_json()})


@app.route('/user/pwd', methods=['PUT'])
@login_required
def putUserPWD():
    if not request.json or not 'current_pwd' in request.json or not 'new_pwd' in request.json:
        return jsonify({'err': 'Request not Json or miss current_pwd/new_pwd'})
    else:
        current_pwd = current_user.pwd
    if not request.json['current_pwd'] == current_pwd:
        return jsonify({'err': 'current_pwd is not right.'})
    else:
        current_user.pwd = request.json['new_pwd']
        try:
            current_user.save()
        except Exception:
            return jsonify({'err': 'Modify PWD error.'})
        return jsonify({'status': 0, 'msg': 'PWD has been modified.', 'user_id': current_user.user_id})


if __name__ == '__main__':
    app.run(debug=True)
「原创声明：保留所有权利，禁止转载」
2 个赞
共收到 1 条回复 时间 点赞
mark #2 · 2018年10月05日

老哥，能给份源码吗

需要 登录 后方可回复, 如果你还没有账号请点击这里 注册。