Spaces:
Sleeping
Sleeping
File size: 2,894 Bytes
d74c0dc |
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 |
#!/usr/bin/env python3
"""Test Supabase connection directly"""
import os
from dotenv import load_dotenv
load_dotenv()
try:
from supabase import create_client
supabase_url = os.getenv("SUPABASE_URL")
supabase_key = os.getenv("SUPABASE_SERVICE_KEY")
print("Testing Supabase Connection:")
print("=" * 50)
print(f"URL: {supabase_url}")
print(f"Key length: {len(supabase_key) if supabase_key else 0}")
print()
if not supabase_url or not supabase_key:
print("ERROR: Missing Supabase credentials")
exit(1)
print("Creating Supabase client...")
client = create_client(supabase_url, supabase_key)
print("[OK] Client created successfully")
print()
print("Testing table access...")
tables = ["tool_usage_events", "redflag_violations", "rag_search_events", "agent_query_events"]
for table in tables:
try:
result = client.table(table).select("id").limit(1).execute()
print(f"[OK] Table '{table}' is accessible")
except Exception as e:
error_msg = str(e)
if "does not exist" in error_msg.lower() or "relation" in error_msg.lower():
print(f"[ERROR] Table '{table}' does NOT exist")
print(f" Solution: Run supabase_analytics_tables.sql in Supabase SQL Editor")
elif "401" in error_msg or "Invalid API key" in error_msg:
print(f"[ERROR] Table '{table}' access denied - Invalid API key")
print(f" Error: {error_msg[:100]}")
else:
print(f"[ERROR] Table '{table}' error: {error_msg[:100]}")
print()
print("Testing insert...")
try:
test_payload = {
"tenant_id": "test_connection",
"tool_name": "connection_test",
"timestamp": 1234567890,
"success": True
}
result = client.table("tool_usage_events").insert(test_payload).execute()
print("[OK] Test insert successful!")
print(f" Inserted {len(result.data) if result.data else 1} row(s)")
except Exception as e:
error_msg = str(e)
print(f"[ERROR] Test insert failed: {error_msg[:200]}")
if "401" in error_msg or "Invalid API key" in error_msg:
print(" This indicates an invalid API key")
elif "does not exist" in error_msg.lower():
print(" This indicates the table doesn't exist")
elif "RLS" in error_msg or "policy" in error_msg.lower():
print(" This indicates RLS policy blocking the insert")
print()
print("=" * 50)
print("Connection test complete!")
except ImportError:
print("ERROR: supabase-py package not installed")
print("Install it with: pip install supabase")
except Exception as e:
print(f"ERROR: {e}")
import traceback
traceback.print_exc()
|