Coverage for app / core / security.py: 100%

1import asyncio 

2from uuid import UUID 

3from fastapi import HTTPException, status 

4from supabase import create_client 

5 

6from app.core.config import settings 

7 

8_supabase = None 

9 

10def get_supabase_admin(): 

11 global _supabase 

12 if _supabase is None: 

13 _supabase = create_client(settings.supabase_url, settings.supabase_service_key) 

14 return _supabase 

15 

16 

17async def verify_token(token: str) -> dict: 

18 """Verify a Supabase JWT without blocking the event loop.""" 

19 supabase = get_supabase_admin() 

20 try: 

21 response = await asyncio.to_thread(supabase.auth.get_user, token) 

22 if response.user is None: 

23 raise ValueError("Invalid user") 

24 return { 

25 "id": UUID(response.user.id), # ← coerce to UUID 

26 "email": response.user.email, 

27 "supabase_uid": response.user.id, 

28 } 

29 except Exception as e: 

30 raise HTTPException( 

31 status_code=status.HTTP_401_UNAUTHORIZED, 

32 detail=f"Invalid token: {e}", 

33 headers={"WWW-Authenticate": "Bearer"}, 

34 )