使用Flask開發簡單接口(4)--藉助Redis實現token驗證

前言

在以前咱們已開發了幾個接口，而且能夠正常使用，那麼今天咱們將繼續完善一下。咱們注意到以前的接口，都是不須要進行任何驗證就可使用的，其實咱們可使用 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