Spaces:
Running
Running
| """ | |
| Supabase singleton clients. | |
| Two clients: | |
| get_anon_client() — anon key, used for user-facing auth (sign in/out) | |
| get_service_client() — service_role key, used server-side for DB + Storage | |
| (bypasses RLS, never expose to the browser) | |
| get_owner_user_id() — returns the single owner user's UUID (single-user app). | |
| Cached after first call. | |
| """ | |
| import os | |
| SUPABASE_URL = os.getenv("SUPABASE_URL", "") | |
| SUPABASE_ANON_KEY = os.getenv("SUPABASE_ANON_KEY", "") | |
| SUPABASE_SERVICE_ROLE_KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY", "") | |
| _anon_client = None | |
| _service_client = None | |
| _owner_user_id = None # cached after first admin lookup | |
| def is_configured() -> bool: | |
| return bool(SUPABASE_URL and SUPABASE_ANON_KEY) | |
| def get_anon_client(): | |
| global _anon_client | |
| if _anon_client is None: | |
| from supabase import create_client | |
| _anon_client = create_client(SUPABASE_URL, SUPABASE_ANON_KEY) | |
| return _anon_client | |
| def get_service_client(): | |
| global _service_client | |
| if _service_client is None: | |
| from supabase import create_client | |
| _service_client = create_client(SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY) | |
| return _service_client | |
| def get_owner_user_id() -> str | None: | |
| """Return the single owner user's UUID. Cached after first successful lookup.""" | |
| global _owner_user_id | |
| if _owner_user_id: | |
| return _owner_user_id | |
| try: | |
| sb = get_service_client() | |
| result = sb.auth.admin.list_users() | |
| users = result if isinstance(result, list) else getattr(result, "users", []) | |
| if users: | |
| _owner_user_id = users[0].id | |
| return _owner_user_id | |
| except Exception: | |
| pass | |
| return None | |