| | import os |
| | from dotenv import load_dotenv |
| | from supabase import create_client, Client |
| | from supabase_auth.errors import AuthApiError |
| |
|
| | |
| | load_dotenv() |
| |
|
| | |
| | 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: |
| | |
| | |
| | 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: |
| | |
| | users_list = supabase_admin.auth.admin.list_users() |
| | |
| | if users_list: |
| | 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() |
| |
|
| | |
| | test_email = "test@example.com" |
| | test_password = "testpassword" |
| | await test_user_registration(test_email, test_password) |
| | |
| | await list_supabase_users() |
| |
|
| | if __name__ == "__main__": |
| | import asyncio |
| | asyncio.run(main()) |
| |
|