Spaces:

airsltd
/

mcps

Sleeping

mcps / test /test_sb_create_table.py

update

4469050 over 1 year ago

4.45 kB

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 测试完成。")

	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 测试完成。")