File size: 3,688 Bytes
133609a |
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 |
import os
from dotenv import load_dotenv
from supabase import create_client, Client
from supabase_auth.errors import AuthApiError
# Load environment variables from .env file
load_dotenv()
# Supabase Configuration
SUPABASE_URL = os.environ.get("SUPABASE_URL")
SUPABASE_ANON_KEY = os.environ.get("SUPABASE_ANON_KEY")
if not SUPABASE_URL or not SUPABASE_ANON_KEY:
print("Error: Supabase URL and Anon Key must be set as environment variables in .env file.")
exit(1)
supabase_anon: Client = create_client(SUPABASE_URL, SUPABASE_ANON_KEY)
SUPABASE_SERVICE_ROLE_KEY = os.environ.get("SUPABASE_SERVICE_ROLE_KEY")
if not SUPABASE_SERVICE_ROLE_KEY:
print("Error: SUPABASE_SERVICE_ROLE_KEY must be set as an environment variable in .env file for admin operations.")
exit(1)
supabase_admin: Client = create_client(SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY)
async def test_db_connection():
"""Tests a simple database connection by querying a non-existent table."""
print("\n--- Testing Database Connection ---")
try:
# Attempt a simple query to check connection.
# This will likely fail if the table doesn't exist, but confirms connection.
res = supabase_anon.table('non_existent_table').select('*').limit(1).execute()
print("Database connection successful (or query executed without connection error).")
print(f"Query result (expected empty or error): {res.data}")
except Exception as e:
print(f"Database connection test failed: {e}")
print("Please ensure SUPABASE_URL is correct and your network allows access to Supabase.")
async def test_user_registration(email, password):
"""Tests user registration with Supabase."""
print(f"\n--- Testing User Registration for {email} ---")
try:
res = supabase_anon.auth.sign_up({
"email": email,
"password": password
})
if res.user:
print(f"User registration successful for {res.user.email}! Please check your email for verification.")
print(f"User ID: {res.user.id}")
else:
print("User registration failed: No user returned.")
print(f"Supabase response: {res}")
except AuthApiError as e:
print(f"User registration failed (AuthApiError): {e.message}")
except Exception as e:
print(f"An unexpected error occurred during registration: {e}")
print("Please ensure Supabase URL and Anon Key are correct, and Email Signups are enabled in Supabase Auth settings.")
async def list_supabase_users():
"""Lists all users in Supabase."""
print("\n--- Listing Supabase Users ---")
try:
# Supabase admin client is required to list users
users_list = supabase_admin.auth.admin.list_users()
if users_list: # Assuming it returns a list of user objects directly
print(f"Found {len(users_list)} users:")
for user in users_list:
print(f" User ID: {user.id}, Email: {user.email}, Created At: {user.created_at}")
else:
print("No users found in Supabase.")
except Exception as e:
print(f"Failed to list Supabase users: {e}")
print("Please ensure your Supabase Anon Key has sufficient permissions or use a service role key if necessary.")
async def main():
await test_db_connection()
# Replace with a test email and password
test_email = "test@example.com" # Using a more standard test email format
test_password = "testpassword"
await test_user_registration(test_email, test_password)
await list_supabase_users()
if __name__ == "__main__":
import asyncio
asyncio.run(main())
|