Flask 開發(fā)（第二章）--- 用戶憑證

第一章 Flask安裝&網(wǎng)站提供API實例 無授權(quán)直接提供服務(wù)
第二章 用戶憑證 授權(quán)才可訪問API(文件授權(quán)&數(shù)據(jù)庫授權(quán)）

7. 基于文件授權(quán)

項目里創(chuàng)建一個文件 db.txt

為每一個客戶生成uuid。比如使用 python標(biāo)準(zhǔn)庫uuid。

就得到UUID 8580dd0d-00d0-4c48-8a3f-c3d0b41039b2

客戶信息和uuid等信息 寫入 db.txt

image.png

7.1 服務(wù)器源碼中校驗授權(quán)

為了授權(quán)，所做的改動

7.1.1 完整源碼

db.txt

8580dd0d-00d0-4c48-8a3f-c3d0b41039b2,李陽
31ce3f9e-ec0c-4a4b-806c-359da45223e6,張三

main.py

from flask import Flask,request,jsonify
import hashlib

app = Flask(__name__)

def get_user_dict():
    info_dict = {}
    with open("db.txt",mode='r',encoding='utf-8') as f:
        for line in f:
            line = line.strip()
            token,name = line.split(",")
            info_dict[token] = name
    return info_dict

@app.route("/bili", methods = ['POST'])  # 設(shè)置此API只支持POST
def bili():
    """
        請求的URL里 要求攜帶token 比如 /bili?token=8580dd0d-00d0-4c48-8a3f-c3d0b41039b2
        請求的數(shù)據(jù) 格式要求: { "ordered_string":"......" }
    """
    token = request.args.get("token")
    if not token :
        return jsonify({'status':False,'error':"認(rèn)證失敗"})

    user_dict = get_user_dict()
    if token not in user_dict:
        return jsonify({'status':False,'error':"認(rèn)證失敗"})

    ordered_string = request.json.get("ordered_string")
    if not ordered_string :
        return jsonify({'status':False,'error':"參數(shù)錯誤"})

    # 調(diào)用核心算法生成sign簽名
    encrypted_string = ordered_string + "560c52ccd288fed04585"
    obj = hashlib.md5(encrypted_string.encode('utf-8'))
    sign = obj.hexdigest()
    return jsonify({'status':True,'data':sign})

if __name__ == '__main__':
    app.run(host="127.0.0.1",port=5000) # 無參調(diào)用時就是 默認(rèn)用這ip和端口號

7.1.2 瀏覽器端效果演示

  • URL中無token的情況，認(rèn)證失敗

  • URL中有token 但是token不正確的情況，認(rèn)證失敗

  • URL中有正確token，授權(quán)成功，執(zhí)行算法并返回結(jié)果

為什么 基于文件？而不是直接在服務(wù)器源碼里 寫一個dict對象來記錄？（如下圖：）

寫死在源碼里的dict對象

答：來了一個新的客戶，或者客戶信息變動，修改服務(wù)器源碼還得重啟服務(wù)器。每次從文件去獲取的形式，可以直接修改文件 db.txt，更新就成功了。不用依賴重啟服務(wù)器程序來生效。

8.基于數(shù)據(jù)庫授權(quán)

8.1創(chuàng)建數(shù)據(jù)庫

8.2 創(chuàng)建用戶表

8.3 連接數(shù)據(jù)庫

8.3.1 安裝 pymysql

8.3.2 連接mysql數(shù)據(jù)庫 做身份認(rèn)證

調(diào)用和傳參處

與mysql通信的具體實現(xiàn)(性能損耗嚴(yán)重版，連接是屢建屢斷)

一般不允許這樣 連接建立好僅僅執(zhí)行一個語句就關(guān)閉。資源浪費。

數(shù)據(jù)庫為空的狀態(tài)
手動給數(shù)據(jù)庫填上此用戶
數(shù)據(jù)庫中有此token則成功了

8.4 完整源碼

from flask import Flask,request,jsonify
import hashlib
import pymysql

def fetchone(sql_cmd,args):
    conn = pymysql.connect(host='127.0.0.1',port=3306,user='root',passwd='genius',charset='utf8',db='day20')
    cursor = conn.cursor()
    cursor.execute(sql_cmd,args)
    result = cursor.fetchone()
    cursor.close()
    conn.close()
    return result

app = Flask(__name__)

@app.route("/bili", methods = ['POST'])  # 設(shè)置此API只支持POST
def bili():
    """
        請求的URL里 要求攜帶token 比如 /bili?token=8580dd0d-00d0-4c48-8a3f-c3d0b41039b2
        請求的數(shù)據(jù) 格式要求: { "ordered_string":"......" }
    """
    # 1. check token if it's empty
    token = request.args.get("token")
    if not token :
        return jsonify({'status':False,'error':"認(rèn)證失敗"})

    # 2. check token if it's leagal.  link MySql to execute command.
    result = fetchone("select * from day20.my_users where token = %s", [token])
    if not result:
        return jsonify({'status':False,'error':"認(rèn)證失敗"})

    ordered_string = request.json.get("ordered_string")
    if not ordered_string :
        return jsonify({'status':False,'error':"參數(shù)錯誤"})

    # 調(diào)用核心算法生成sign簽名
    encrypted_string = ordered_string + "560c52ccd288fed04585"
    obj = hashlib.md5(encrypted_string.encode('utf-8'))
    sign = obj.hexdigest()
    return jsonify({'status':True,'data':sign})

if __name__ == '__main__':
    app.run(host="127.0.0.1",port=5000) # 無參調(diào)用時就是 默認(rèn)用這ip和端口號

9.（避免性能浪費，使用）數(shù)據(jù)庫的連接池子

a.若每次請求做一個連接建立和連接斷開。數(shù)據(jù)的傳輸和處理時間很短，大比重時間耗費在連接建立和斷開。
b. 若僅使用同一個連接，那就需要考慮并發(fā)問題，并發(fā)性不大高。
c. 連接池方案：一開始運行服務(wù)器就創(chuàng)建多個連接保持存活，來了一個請求，就由空閑的連接來服務(wù)此請求。

9.1 安裝dbutils

9.2 生成連接池 以及 應(yīng)用連接池

9.3 效果演示：和之前一樣，只是效率更高

9.4 完整源碼

from flask import Flask,request,jsonify
import hashlib
import pymysql

from dbutils.pooled_db import PooledDB

POOL = PooledDB(
    creator=pymysql, # select database module
    maxconnections=10, # connection pool capacity, 0 and None means no limit
    mincached=2, # smallest number of the created free connections, 0 means no created
    maxcached=3, # largest number of the created free connections, 0/None means no limit
    blocking=True, # block waiting or no,  when there is no free connection
    setsession=[], # before-session excution list
    ping=0, #
    host='127.0.0.1',port=3306,user='root',passwd='genius',charset='utf8',db='day20'
)

def fetchone(sql_cmd,args):
    # conn = pymysql.connect(host='127.0.0.1',port=3306,user='root',passwd='genius',charset='utf8',db='day20')
    conn = POOL.connection()
    cursor = conn.cursor()
    cursor.execute(sql_cmd,args)
    result = cursor.fetchone()
    cursor.close()
    conn.close()
    return result

app = Flask(__name__)

@app.route("/bili", methods = ['POST'])  # 設(shè)置此API只支持POST
def bili():
    """
        請求的URL里 要求攜帶token 比如 /bili?token=8580dd0d-00d0-4c48-8a3f-c3d0b41039b2
        請求的數(shù)據(jù) 格式要求: { "ordered_string":"......" }
    """
    # 1. check token if it's empty
    token = request.args.get("token")
    if not token :
        return jsonify({'status':False,'error':"認(rèn)證失敗"})

    # 2. check token if it's leagal.  link MySql to execute command.
    result = fetchone("select * from day20.my_users where token = %s", [token])
    if not result:
        return jsonify({'status':False,'error':"認(rèn)證失敗"})

    ordered_string = request.json.get("ordered_string")
    if not ordered_string :
        return jsonify({'status':False,'error':"參數(shù)錯誤"})

    # 調(diào)用核心算法生成sign簽名
    encrypted_string = ordered_string + "560c52ccd288fed04585"
    obj = hashlib.md5(encrypted_string.encode('utf-8'))
    sign = obj.hexdigest()
    return jsonify({'status':True,'data':sign})

if __name__ == '__main__':
    app.run(host="127.0.0.1",port=5000) # 無參調(diào)用時就是 默認(rèn)用這ip和端口號
最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容