[Python] 纯文本查看 复制代码import time
import requests
from retrying import retry
# 定义 API 服务的基本 URL
BASE_URL = 'http://127.0.0.1:5000'
access_token = str()
refresh_token = str()
def login(uname: str, passwd: str):
login_url = f"{BASE_URL}/login"
data = {"username": uname, "password": passwd}
resp = requests.post(login_url, json=data)
if resp.status_code == 200:
return resp.json().get("access_token"), resp.json().get("refresh_token")
else:
print("Login failed.")
return None, None
@retry(stop_max_attempt_number=3, wait_fixed=2000)
def refresh_access_token():
global access_token
refresh_url = f"{BASE_URL}/refresh"
headers = {
"Authorization": f"Bearer {refresh_token}"
}
resp = requests.post(refresh_url, headers=headers)
if resp.status_code == 200:
access_token = resp.json().get("access_token")
return True
else:
print("Token refresh failed.")
return None
def with_refresh(func):
def wrapper(*arg, **kwargs):
try:
resp = func(*arg, **kwargs)
if resp.status_code == 401:
raise Exception(resp.json().get('msg'))
else:
return resp
except Exception as e:
print(f"Error: {e}")
new_access_token = refresh_access_token()
if new_access_token:
return func(*arg, **kwargs)
else:
print("Token refresh failed.")
return None
return wrapper
@with_refresh
def get_protected_data():
headers = {"Authorization": f"Bearer {access_token}"}
protected_url = f"{BASE_URL}/data"
return requests.get(protected_url, headers=headers)
if __name__ == "__main__":
access_token, refresh_token = login("test", "test")
response = get_protected_data()
print(response.json())
time.sleep(62)
response = get_protected_data()
print(response.json())
