Spaces:
Sleeping
Sleeping
File size: 3,824 Bytes
8abdadf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | import os
import requests
from dotenv import load_dotenv
# 加载 .env 文件中的环境变量
load_dotenv()
UPSTASH_REDIS_REST_URL = os.getenv("UPSTASH_REDIS_REST_URL")
UPSTASH_REDIS_REST_TOKEN = os.getenv("UPSTASH_REDIS_REST_TOKEN")
if not UPSTASH_REDIS_REST_URL or not UPSTASH_REDIS_REST_TOKEN:
print("错误：UPSTASH_REDIS_REST_URL 或 UPSTASH_REDIS_REST_TOKEN 未设置。请检查 .env 文件。")
exit(1)
headers = {
"Authorization": f"Bearer {UPSTASH_REDIS_REST_TOKEN}",
"Content-Type": "application/json"
}
def execute_redis_command(command, *args):
url = ""
send_json_payload = False
payload = []
if command.upper() == "SET":
if len(args) != 2:
print("SET 命令需要两个参数：key 和 value。")
return None
key, value = args
url = f"{UPSTASH_REDIS_REST_URL}/set/{key}/{value}"
elif command.upper() == "GET":
if len(args) != 1:
print("GET 命令需要一个参数：key。")
return None
key = args[0]
url = f"{UPSTASH_REDIS_REST_URL}/get/{key}"
elif command.upper() == "DEL":
if len(args) != 1:
print("DEL 命令需要一个参数：key。")
return None
key = args[0]
url = f"{UPSTASH_REDIS_REST_URL}/del/{key}"
else:
# 对于其他命令，使用通用格式，并发送 JSON 数组作为 payload
url = f"{UPSTASH_REDIS_REST_URL}/{command}"
payload = [arg for arg in args]
send_json_payload = True
try:
if send_json_payload:
response = requests.post(url, headers=headers, json=payload)
else:
response = requests.post(url, headers=headers) # 不发送 json=payload
response.raise_for_status() # 如果请求失败，抛出 HTTPError
return response.json()
except requests.exceptions.HTTPError as http_err:
print(f"HTTP 错误发生: {http_err}")
print(f"响应内容: {response.text}")
except requests.exceptions.ConnectionError as conn_err:
print(f"连接错误发生: {conn_err}")
except requests.exceptions.Timeout as timeout_err:
print(f"请求超时: {timeout_err}")
except requests.exceptions.RequestException as req_err:
print(f"发生未知错误: {req_err}")
return None
def test_redis_connection():
print("正在测试 Upstash Redis 连接...")
# SET 命令
key = "mykey"
value = "hello_upstash"
print(f"正在设置键 '{key}' 为值 '{value}'...")
set_result = execute_redis_command("SET", key, value)
if set_result:
print(f"SET 结果: {set_result}")
if set_result and set_result.get('result') == "OK":
print("SET 命令成功。")
else:
print("SET 命令失败。")
return # 如果 SET 失败，才返回
# GET 命令
print(f"正在获取键 '{key}' 的值...")
get_result = execute_redis_command("GET", key)
if get_result is not None:
print(f"GET 结果: {get_result}")
if get_result and get_result.get('result') == value:
print("GET 命令成功，值匹配。")
else:
print("GET 命令失败，值不匹配或结果格式不正确。")
else:
print("GET 命令失败。")
return
# DEL 命令 (可选，用于清理)
print(f"正在删除键 '{key}'...")
del_result = execute_redis_command("DEL", key)
if del_result is not None:
print(f"DEL 结果: {del_result}")
if del_result and del_result.get('result') == 1:
print("DEL 命令成功。")
else:
print("DEL 命令失败或结果格式不正确。")
else:
print("DEL 命令失败。")
if __name__ == "__main__":
test_redis_connection()
|