File size: 4,454 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 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 111 112 113 114 |
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 create_test_table():
"""
创建 Supabase 中的 'test_table_1' 表。
如果表已存在，则不会重复创建。
"""
table_name = "test_table_1"
print(f"正在尝试通过 RPC 创建表 '{table_name}'...")
# Supabase Python SDK 不直接支持执行任意 DDL SQL。
# 最推荐的方式是在 Supabase 控制台手动创建表，或者通过迁移工具。
# 如果必须通过代码执行 DDL，通常需要：
# 1. 在 Supabase 数据库中创建一个 PostgreSQL 函数来封装 DDL 逻辑。
# 2. 然后通过 supabase.rpc() 调用这个函数。
# 以下是需要在 Supabase SQL 编辑器中执行的 PostgreSQL 函数定义：
print("\n--- 请在 Supabase SQL 编辑器中执行以下 SQL 来创建函数 'create_table_if_not_exists_func' ---")
print(f"""
CREATE OR REPLACE FUNCTION create_table_if_not_exists_func(table_name TEXT)
RETURNS VOID AS $$
BEGIN
EXECUTE format('
CREATE TABLE IF NOT EXISTS %I (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
name TEXT,
age INT
);
', table_name);
END;
$$ LANGUAGE plpgsql;
""")
print("----------------------------------------------------------------------------------\n")
# 尝试通过 RPC 调用上述 PostgreSQL 函数
try:
response = supabase.rpc('create_table_if_not_exists_func', {'table_name': table_name}).execute()
if response.error:
print(f"通过 RPC 调用函数创建表失败: {response.error}")
# 如果错误是由于函数不存在，提示用户先创建函数
if "function create_table_if_not_exists_func(text) does not exist" in str(response.error):
print("错误提示：PostgreSQL 函数 'create_table_if_not_exists_func' 不存在。请先在 Supabase 控制台创建该函数。")
else:
print(f"通过 RPC 调用函数创建表成功: {response.data}")
except Exception as e:
print(f"调用 RPC 函数时发生意外错误: {e}")
def test_insert_and_query_data():
"""
测试向 Supabase 插入数据并查询。
请确保你的 Supabase 数据库中存在一个名为 'test_table_1' 的表，
并且该表包含 'name' (text) 和 'age' (integer) 字段。
"""
table_name = "test_table_1" # 替换为你的表名
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 测试...")
create_test_table() # 在运行测试之前尝试创建表
test_insert_and_query_data() # 取消注释，以便在创建表后运行数据操作测试
print("Supabase 测试完成。")
|