使用令牌认证和 functools.wraps 构建装饰器

Question

我正在构建一个执行 API 请求的 kivy 应用程序。 get_token() 函数在启动应用程序时首先运行以获取所有必要的信息、字典、变量等。在令牌过期 30 分钟后，应用程序有一个按钮，on_press 它将向服务器发送请求调用，如果令牌过期它不会工作。我正在尝试构建一个装饰器@use_token，它将检查简单请求是否被执行并接收响应 200 成功代码，如果成功：返回（装饰函数）否则：调用 get_token 将刷新令牌，并返回装饰函数。整个概念有效，打印在每个函数中每个可能的代码块上工作，但是当 30 分钟令牌过期后，refresh_token() 在 if 语句中调用 get_token()，打印仍然有效，但返回的修饰函数不发送请求，它打印测试字符串但不做主要的事情。这告诉我函数 get_token() 在 if 语句中执行但不更新令牌信息...

第一个函数，在开始时运行一个：

def get_token():
    url_token = "http://server.com"
    payload = "{" \
              "\n  \"grantType\": \"password\"," \
              "\n  \"password\": \"string\"," \
              "\n  \"refreshToken\": \"string\"," \
              "\n  \"token\": \"string\"," \
              "\n  \"username\": \"admin\"" \
              "\n}"
    headers = {
            'Content-Type': 'application/json',
            'api_key': ''
        }
    global readyToken
    readyToken = requests.request("POST", url_token, headers=headers, data=payload).json()['token']
    print("Getting a NEW TOKEN!")
get_token()

装饰功能：

def use_token(func):
    @functools.wraps(func)
    def refresh_token(*args):
        url_check = "simplerequest.com"
        response = requests.request("PUT", url_check, headers=HEADERS)
        print("This print from url_check block "+str(response))
        str_response = str(response)
        if '401' in str_response:
            print("401 found, Token is Expired, refreshing with get_token")
            get_token()
        else:
            print("200 Code, success, passing, leaving else statement")
            pass
        print("emptying str_response and calling for decorated function:")
        str_response = ""
        return func(*args)
    return refresh_token

装饰功能：

global URL_QC, HEADERS

URL_QC = "www.server.com"
HEADERS = {
        'Content-Type': 'application/json',
        'api_key': readyToken
}


@use_token
def change_channel(self, display_mac, ch_number):
    print("Hello from DECORATED function!!!")
    payload = "{\"deviceIds\": [" + str(display_mac) + "],\"menu\": \"save_sch_channel\", \"productType\": \"string\", \"value\":" + str(ch_number) + "}"
    response = requests.request("PUT", URL_QC, headers=HEADERS, data=payload)

Answer 1

readyToken 更改时，您并未更新 HEADERS 变量。

HEADERS = {
  'Content-Type': 'application/json',
  'api_key': readyToken
}

在python中，字符串是按值传递的。所以在这里，您只需将HEADERS['api_key'] 设置为readyToken 的当前值一次。如果稍后更改readyToken，HEADERS 不会更新，因为它只保留readyToken 的原始值。

这可以通过在HEADERS 每次更改时简单地更新readyToken 来解决：

def get_token():
    url_token = "http://server.com"
    payload = {
        "grantType": "password",
        "password":  "string",
        "refreshToken": "string",
        "token": "string",
        "username": "admin"
    }
    headers = {
        "Content-Type": 'application/json',
        "api_key": ""
    }
    response = requests.post(url_token, headers=headers, data=payload)
    print(f'Got Response: {response.json()}')
    global readyToken
    readyToken = response.json()['token']

    # We also need to update headers!
    global HEADERS
    HEADERS['api_key'] = readyToken

更好的是，如果您只在 HEADERS 中使用 readyToken，请完全删除 readyToken 变量并简单地更新全局 HEADERS 变量：

def get_token():
    ...
    response = requests.post(url_token, headers=headers, data=payload)
    print(f'Got Response: {response.json()}')
    global HEADERS
    HEADERS['api_key'] = response.json()['token']

---

Answer 2

Alfa Q 非常感谢您，您的所有猜测和修正都是正确的！我确实只在标头中使用令牌，因此我读取了 readyToken 并将令牌表达式直接放入“api_key”值中。我还重组了代码并创建了一个实现所有修复的类：

token.py：

class Token:
    def __init__(self):
        self.url_check = "www.check.com"
        self.url_token = "www.server.com"
        self.payload = "{" \
                  "\n  \"grantType\": \"password\"," \
                  "\n  \"password\": \"password\"," \
                  "\n  \"refreshToken\": \"string\"," \
                  "\n  \"token\": \"string\"," \
                  "\n  \"username\": \"admin\"" \
                  "\n}"
        self.headers_token = {
                'Content-Type': 'application/json',
                'api_key': self.get_token()
            }

    def get_token(self):
        return requests.post(self.url_token, headers=self.headers, data=self.payload).json()['token']

    def use_token(self, func):
        @functools.wraps(func)
        def refresh_token(*args):
            response = requests.put(self.url_check, headers=self.headers_token)
            print("This print from url_check block: " + str(response))
            if response.status_code == 200:
                print('Status 200 | Token OK - No refresh necessary')
                return func(*args)
            elif response.status_code == 401:
                print('Status 401 | Token is Expired - Refreshing')
                self.get_token()
                return func(*args)
            else:
                print(f'Status {response.status_code} | Error Occurred')
                print("Print from REFRESH_TOKEN")
        return refresh_token

main.py:

@token.use_token
def change_channel(self, display_mac, ch_number):
    print("Hello from DECORATED function!!!")
    payload = "{\"deviceIds\": [" + str(display_mac) + "],\"menu\": \"save_sch_channel\", \"productType\": \"string\", \"value\":" + str(ch_number) + "}"
    response = requests.request("PUT", URL_QC, headers=token.headers_token, data=payload)
    print(response)