File size: 2,243 Bytes
4469050 |
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 |
from dotenv import load_dotenv # 导入 load_dotenv
# 加载 .env 文件中的环境变量
load_dotenv()
from supabase import create_client, Client
import os
# 从环境变量获取 Supabase URL 和 Key
# 在实际应用中，请确保这些变量已设置
supabase_url = os.getenv("SUPABASE_URL")
supabase_key = os.getenv("SUPABASE_KEY")
# !!! 请替换为你的 Supabase 项目 URL 和 anon key !!!
# SUPABASE_URL = "SUPABASE_URL"
# SUPABASE_KEY = "SUPABASE_KEY"
# 创建 Supabase 客户端
supabase: Client = create_client(supabase_url, supabase_key)
def test_insert_and_query_data():
"""
测试向 Supabase 插入数据并查询。
请确保你的 Supabase 数据库中存在一个名为 'test_table' 的表，
并且该表包含 'name' (text) 和 'age' (integer) 字段。
"""
table_name = "test_table" # 替换为你的表名
print(f"正在向表 '{table_name}' 插入数据...")
# 插入数据
data_to_insert = {"name": "Test User", "age": 30}
response = supabase.table(table_name).insert(data_to_insert).execute()
if response.data:
print(f"数据插入成功: {response.data}")
else:
print(f"数据插入失败: {response.error}")
assert False, f"插入失败: {response.error}"
print(f"正在从表 '{table_name}' 查询数据...")
# 查询数据
response = supabase.table(table_name).select("*").eq("name", "Test User").execute()
if response.data:
print(f"数据查询成功: {response.data}")
assert len(response.data) > 0, "未查询到数据"
assert response.data[0]["name"] == "Test User"
else:
print(f"数据查询失败: {response.error}")
assert False, f"查询失败: {response.error}"
# 清理数据 (可选，根据测试需求决定是否保留)
print(f"正在从表 '{table_name}' 删除数据...")
response = supabase.table(table_name).delete().eq("name", "Test User").execute()
if response.data:
print(f"数据删除成功: {response.data}")
else:
print(f"数据删除失败: {response.error}")
if __name__ == "__main__":
print("开始运行 Supabase 测试...")
test_insert_and_query_data()
print("Supabase 测试完成。")
|