test/test_sb.py

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