使用Flask开发简单接口(4)--借助Redis实现token验证

2021-10-11 02:15:03

前言

在之前我们已开发了几个接口，并且可以正常使用，那么今天我们将继续完善一下。我们注意到之前的接口，都是不需要进行任何验证就可以使用的，其实我们可以使用 token ，比如设置在修改或删除用户信息的时候需要进行 token 登录验证，这个地方我们将引入 Redis 用于存储登录时产生的 token 。

本人环境：Python 3.7.0 、Redis 5.0.8

新增Redis配置

我们在项目根路径下 config 包中，修改文件 setting.py ，在该文件中配置 Redis 的服务器地址、端口、密码、token过期时间（单位：秒）等参数。

# Redis配置

REDIS_HOST = "192.168.89.128"

REDIS_PORT = 6379

REDIS_PASSWD = "123456"

# token过期时间（单位：秒）

EXPIRE_TIME = 600

Python操作Redis

安装redis模块

使用Python来操作Redis，需要用到 redis 这个第三方库，具体安装方法如下：

pip install redis

我这里安装的版本是 3.4.1。

D:\>pip3 show redis

Name: redis

Version: 3.4.1

Summary: Python client for Redis key-value store

Home-page: https://github.com/andymccurdy/redis-py

Author: Andy McCurdy

Author-email: sedrik@gmail.com

License: MIT

Location: d:\python\installation\lib\site-packages

Requires:

Required-by:

封装Python操作Redis的代码

我们在项目根路径下 common 包中，新建文件 redis_operate.py ，该文件下简单封装了Python操作Redis的代码，后续将通过调用该文件的 redis_db 对象及方法来操作Redis。

import redis

from config.setting import REDIS_HOST, REDIS_PORT, REDIS_PASSWD, EXPIRE_TIME

class RedisDb():

    def __init__(self, host, port, passwd):

        # 建立数据库连接

        self.r = redis.Redis(

            host=host,

            port=port,

            password=passwd,

            decode_responses=True # get() 得到字符串类型的数据

        )

    def handle_redis_token(self, key, value=None):

        if value: # 如果value非空，那么就设置key和value，EXPIRE_TIME为过期时间

            self.r.set(key, value, ex=EXPIRE_TIME)

        else: # 如果value为空，那么直接通过key从redis中取值

            redis_token = self.r.get(key)

            return redis_token

redis_db = RedisDb(REDIS_HOST, REDIS_PORT, REDIS_PASSWD)

在Redis中，我们存储数据是 str 字符串类型，但由于python3与redis交互的驱动问题，默认通过 get() 取出来是 bytes 字节类型，为解决这个问题，我们在上面代码创建连接时，设置了 decode_responses=True ，这样通过 get() 取出来就是 str 字符串类型。

登录时设置token

一般情况，我们登录成功之后，才会产生token，以便在请求其他接口时进行登录验证。因此，我们需要在返回登录成功信息之前，设置token，并把当前登录用户和设置token，作为redis中的 key 和 value ，放到redis中进行存储。修改用户登录接口如下：

@app.route("/login", methods=['POST'])

def user_login():

    """登录用户"""

    username = request.values.get("username", "").strip()

    password = request.values.get("password", "").strip()

    if username and password: # 注意if条件中空串 "" 也是空, 按False处理

        sql1 = "SELECT username FROM user WHERE username = '{}'".format(username)

        res1 = db.select_db(sql1)

        print("查询到用户名 ==>> {}".format(res1))

        if not res1:

            return jsonify({"code": 1003, "msg": "用户名不存在！！！"})

        sql2 = "SELECT * FROM user WHERE username = '{}' and password = '{}'".format(username, password)

        res2 = db.select_db(sql2)

        print("获取 {} 用户信息 == >> {}".format(username, res2))

        if res2:

            timeStamp = int(time.time()) # 获取当前时间戳

            token = "{}{}".format(username, timeStamp)

            redis_db.handle_redis_token(username, token) # 把token放到redis中存储

            login_info = { # 构造一个字典，将 id/username/token/login_time 返回

                "id": res2[0]["id"],

                "username": username,

                "token": token,

                "login_time": time.strftime("%Y/%m/%d %H:%M:%S")

            }

            return jsonify({"code": 0, "login_info": login_info, "msg": "恭喜，登录成功！"})

        return jsonify({"code": 1002, "msg": "用户名或密码错误！！！"})

    else:

        return jsonify({"code": 1001, "msg": "用户名或密码不能为空！！！"})

在这里，我们设置的 token 是由 用户名+当前时间戳 拼接而成的，并把其放到redis中，可以到redis中查看token，或者在登录成功后接口返回数据中的 login_info 中查看。而 token 的有效时间，这个是在配置文件 setting.py 中设置的 EXPIRE_TIME = 600 ，即 600 秒 后当前用户的 token 才会过期失效。

删除用户请求接口实现

接下来，我们准备新开发个接口：删除用户接口。该接口需要 管理员用户 登录认证后才可以进行操作，管理员用户可以删除任何 普通用户 信息。

原本我是想通过 DELETE 请求方式来开发该接口，但好像使用 DELETE 方式的HTTP请求中，不允许传入Body，因此为了方便一些，我们这里仍使用 POST 方式来请求。

删除用户接口（POST请求方式）

@app.route("/delete/user/<int:id>", methods=['POST'])

def user_delete(id):

    username = request.json.get("username", "").strip()  # 当前登录的管理员用户

    token = request.json.get("token", "").strip()  # token口令

    if username and token:

        redis_token = redis_db.handle_redis_token(username)  # 从redis中取token

        if redis_token:

            if redis_token == token:  # 如果从redis中取到的token不为空，且等于请求body中的token

                sql1 = "SELECT role FROM user WHERE username = '{}'".format(username)

                res1 = db.select_db(sql1)

                print("根据用户名 【 {} 】 查询到用户类型 == >> {}".format(username, res1))

                user_role = res1[0]["role"]

                if user_role == 0:  # 如果当前登录用户是管理员用户

                    sql2 = "SELECT * FROM user WHERE id = '{}'".format(id)

                    res2 = db.select_db(sql2)

                    print("根据用户ID 【 {} 】 查询到用户信息 ==>> {}".format(id, res2))

                    if not res2:  # 如果要删除的用户不存在于数据库中，res2为空

                        return jsonify({"code": 3005, "msg": "删除的用户ID不存在，无法进行删除，请检查！！！"})

                    elif res2[0]["role"] == 0: # 如果要删除的用户是管理员用户，则不允许删除

                        return jsonify({"code": 3006, "msg": "用户ID：【 {} 】，该用户不允许删除！！！".format(id)})

                    else:

                        sql3 = "DELETE FROM user WHERE id = {}".format(id)

                        db.execute_db(sql3)

                        print("删除用户信息SQL ==>> {}".format(sql3))

                        return jsonify({"code": 0, "msg": "恭喜，删除用户信息成功！"})

                else:

                    return jsonify({"code": 3004, "msg": "当前用户不是管理员用户，无法进行操作，请检查！！！"})

            else:

                return jsonify({"code": 3003, "msg": "token口令不正确，请检查！！！"})

        else:

            return jsonify({"code": 3002, "msg": "当前用户未登录，请检查！！！"})

    else:

        return jsonify({"code": 3001, "msg": "管理员用户/token口令不能为空，请检查！！！"})

相关的接口返回码和请求场景如下：

接口返回码	请求场景
0	请求参数正确，删除用户成功！
3001	请求参数中，管理员用户/token口令，任一参数为空
3002	请求参数中，当前操作用户没有token，登录验证失败
3003	请求参数中的token值，与redis中的token值不一致
3004	请求参数中，当前操作用户不是管理员用户，无权限进行操作
3005	请求参数中，要删除的用户ID不存在
3006	请求参数中，要删除的用户是管理员用户，不允许删除

可参考如下进行接口请求：

请求方式：POST

请求地址：http://127.0.0.1:5000/delete/user/5

请求头：

Content-Type: application/json

Body：{"username": "wintest", "token": "wintest1587830406"}

OK，通过以上操作，我们已成功借助Redis来实现token登录验证，当然，我们redis的作用还有很多，在这里我们只是简单的应用了一些，要学习的东西还是很多。相关代码已上传到GitHub，大家有兴趣的可以基于此进行学习及开展接口测试。

GitHub源码地址：https://github.com/wintests/flaskDemo

码农公寓

前言

新增Redis配置

Python操作Redis

登录时设置token

删除用户请求接口实现

相关文章