71 lines
2.3 KiB
Python
71 lines
2.3 KiB
Python
from datetime import datetime, timezone
|
|
from typing import Any, Dict, List, Optional, cast
|
|
|
|
from app.core.supabase import get_supabase
|
|
|
|
|
|
def get_user_by_phone(phone: str) -> Optional[Dict[str, Any]]:
|
|
supabase = get_supabase()
|
|
result = supabase.table("users").select("*").eq("phone", phone).single().execute()
|
|
return cast(Optional[Dict[str, Any]], result.data or None)
|
|
|
|
|
|
def get_user_by_id(user_id: str) -> Optional[Dict[str, Any]]:
|
|
supabase = get_supabase()
|
|
result = supabase.table("users").select("*").eq("id", user_id).single().execute()
|
|
return cast(Optional[Dict[str, Any]], result.data or None)
|
|
|
|
|
|
def user_exists_by_phone(phone: str) -> bool:
|
|
supabase = get_supabase()
|
|
result = supabase.table("users").select("id").eq("phone", phone).execute()
|
|
return bool(result.data)
|
|
|
|
|
|
def create_user(payload: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|
supabase = get_supabase()
|
|
result = supabase.table("users").insert(payload).execute()
|
|
return cast(List[Dict[str, Any]], result.data or [])
|
|
|
|
|
|
def list_users() -> List[Dict[str, Any]]:
|
|
supabase = get_supabase()
|
|
result = supabase.table("users").select("*").order("created_at", desc=True).execute()
|
|
return cast(List[Dict[str, Any]], result.data or [])
|
|
|
|
|
|
def update_user(user_id: str, payload: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|
supabase = get_supabase()
|
|
result = supabase.table("users").update(payload).eq("id", user_id).execute()
|
|
return cast(List[Dict[str, Any]], result.data or [])
|
|
|
|
|
|
def _parse_expires_at(expires_at: Any) -> Optional[datetime]:
|
|
try:
|
|
expires_at_dt = datetime.fromisoformat(str(expires_at).replace("Z", "+00:00"))
|
|
except Exception:
|
|
return None
|
|
|
|
if expires_at_dt.tzinfo is None:
|
|
expires_at_dt = expires_at_dt.replace(tzinfo=timezone.utc)
|
|
return expires_at_dt.astimezone(timezone.utc)
|
|
|
|
|
|
def deactivate_user_if_expired(user: Dict[str, Any]) -> bool:
|
|
expires_at = user.get("expires_at")
|
|
if not expires_at:
|
|
return False
|
|
|
|
expires_at_dt = _parse_expires_at(expires_at)
|
|
if not expires_at_dt:
|
|
return False
|
|
|
|
if datetime.now(timezone.utc) <= expires_at_dt:
|
|
return False
|
|
|
|
user_id = user.get("id")
|
|
if user.get("is_active") and user_id:
|
|
update_user(cast(str, user_id), {"is_active": False})
|
|
|
|
return True
|