引言 最近在编写Flask的过程中，部分接口需要token鉴权操作。于是单独编写一个“token(login)的装饰器”来解决相关token问题。

1.代码实现

1.token装饰器

目的token装饰器：进行部分接口的token鉴权操作。

mport functools
from flask import request, make_response, current_app
from utils.optUtils import traceID
from utils.responseUtil import responseUtil
from functools import wraps
from utils.redisUtil import ops_string
def check_login_status(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        # 如果获取到 session , 即判定为 已登录
        print("ckeck")
        traceId = traceID()
        try:
            requestToken = request.headers['token']
            print("requestToken:\t", requestToken)
            redisOps = ops_string()
            requestTokenSearchInRedis = redisOps.get(requestToken)
            print("requestTokenSearchInRedis:\t", requestTokenSearchInRedis)
            if requestTokenSearchInRedis is None:
                msg = {"code": 500, "msg": "请重新登陆[token错误]"}
                # msg = {"code": 500, "msg": "请重新登陆", "traceId": traceId}
                response = make_response(msg)
                # responseUtil(current_app, logLevel='error', codeFile=codeFile, request=request, requestBody=None,
                #              response=response)
                responseUtil(current_app, logLevel='error', codeFile=None, request=request, requestBody=None,
                             response=response)
                return response, 500
            # if request.cookies:
            #     return func(*args, **kwargs)
            else:
                # 此处是核心
                return func()
        except:
            msg = {"code": 500, "msg": "请重新登陆[缺失token]", "traceId": traceId}
            response = make_response(msg)
            responseUtil(current_app, logLevel='error', request=request, requestBody=None,
                         response=response)
            return response, 500
    return wrapper

2.使用token装饰器

 【更新-计划】
@planApp.post("/update")
@check_login_status # 使用token装饰器
def updatePlan():
    traceId = traceID()
    requestBody = request.get_json()
    print("requestBody:\t", requestBody)
    id = requestBody['id']
    resID = db.session.query(Plan).filter_by(id=id).first()
    print("resID:\t", resID)
    if resID is not None:
        del requestBody['id']  # 不更新id值
        print("处理:\t", requestBody)
        Plan.query.filter_by(id=id).update(requestBody)
        msg = {"code": 0, "msg": "计划更新成功", "traceId": traceId}
        # dictSort(data=msg)  # 按照ascii升序排序
        response = make_response(msg)
        responseUtil(current_app, logLevel='info', request=request,
                     requestBody=requestBody,
                     response=response)
        return response, 200
    else:
        print("计划更新失败")
        msg = {"code": 502, "msg": "计划更新失败", "traceId": traceId}
        response = make_response(msg)
        responseUtil(current_app, logLevel='error', request=request,
                     requestBody=requestBody,
                     response=response)
        return response, 502

3.运行验证

2.【最简示例】flask装饰器

1.代码实现

def check1(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        print("ckeck")
        if 1 == 1:
            # func()
            return func()
        else:
            print("error")
    return wrapper
@planApp.get("/check")
@check1
def hello():
    print("hello-world")
    return "hello check"

2.运行验证