101 lines
5.6 KiB
Python
101 lines
5.6 KiB
Python
|
|
from supabase import create_client, Client
|
|
from supabase.lib.client_options import ClientOptions
|
|
from datetime import datetime, timezone
|
|
import logging
|
|
import traceback
|
|
|
|
class DBManager:
|
|
"""
|
|
DBManager는 Supabase 클라이언트를 래핑하여 로그인, 사용자 정보 조회,
|
|
마지막 로그인 시간 업데이트 등 여러 API 호출을 수행합니다.
|
|
"""
|
|
def __init__(self, logger: logging.Logger):
|
|
self.logger = logger
|
|
self.logger.log("Initializing DBManager...", level=logging.DEBUG)
|
|
# 실제 Supabase URL과 KEY로 수정하세요.
|
|
self.url: str = "http://146.56.101.199:8000"
|
|
self.key: str = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyAgCiAgICAicm9sZSI6ICJhbm9uIiwKICAgICJpc3MiOiAic3VwYWJhc2UtZGVtbyIsCiAgICAiaWF0IjogMTY0MTc2OTIwMCwKICAgICJleHAiOiAxNzk5NTM1NjAwCn0.dc_X5iR_VP_qT0zsiyj_I_OZ2T9FtRU2BBNWN8Bu4GE"
|
|
self.client: Client = create_client(self.url, self.key)
|
|
self.access_token = None
|
|
self.refresh_token = None
|
|
self.logger.log("DBManager initialized", level=logging.DEBUG)
|
|
|
|
def update_client_with_token(self, access_token: str):
|
|
self.logger.log("Entering update_client_with_token()", level=logging.DEBUG)
|
|
try:
|
|
options = ClientOptions(headers={"Authorization": f"Bearer {access_token}"})
|
|
self.client = create_client(self.url, self.key, options=options)
|
|
self.logger.log("Client updated with JWT token", level=logging.DEBUG)
|
|
except Exception as ex:
|
|
self.logger.log(f"update_client_with_token error: {ex}", level=logging.ERROR, exc_info=True)
|
|
self.logger.log("Exiting update_client_with_token()", level=logging.DEBUG)
|
|
|
|
def login(self, email: str, password: str) -> dict:
|
|
self.logger.log("Entering DBManager.login()", level=logging.DEBUG)
|
|
try:
|
|
response = self.client.auth.sign_in_with_password({"email": email, "password": password})
|
|
if response.session:
|
|
if response.user.email_confirmed_at is None:
|
|
self.logger.log("이메일 인증이 완료되지 않았습니다.", level=logging.WARNING)
|
|
return {"error": "이메일 인증을 먼저 완료해 주세요."}
|
|
self.access_token = response.session.access_token
|
|
self.refresh_token = response.session.refresh_token
|
|
self.update_client_with_token(self.access_token)
|
|
user_info = {
|
|
"id": response.user.id,
|
|
"email": response.user.email,
|
|
"nickname": response.user.user_metadata.get("nickname", "Unknown")
|
|
}
|
|
self.logger.log(f"로그인 성공: {user_info}", level=logging.DEBUG)
|
|
self.logger.log("Exiting DBManager.login()", level=logging.DEBUG)
|
|
return user_info
|
|
else:
|
|
self.logger.log("로그인 실패: 세션 없음", level=logging.WARNING)
|
|
return {"error": "로그인 실패"}
|
|
except Exception as e:
|
|
self.logger.log(f"Login error: {e}", level=logging.ERROR, exc_info=True)
|
|
return {"error": str(e)}
|
|
|
|
def get_auth_user_info(self, user_id: str) -> dict:
|
|
self.logger.log(f"Entering get_auth_user_info(user_id={user_id})", level=logging.DEBUG)
|
|
try:
|
|
response = self.client.from_("users").select("email_confirmed_at").eq("id", user_id).execute()
|
|
if response.data and len(response.data) > 0:
|
|
self.logger.log(f"Retrieved auth user info: {response.data[0]}", level=logging.DEBUG)
|
|
return response.data[0]
|
|
else:
|
|
self.logger.log("auth.users에서 사용자 정보를 찾지 못했습니다.", level=logging.WARNING)
|
|
return {}
|
|
except Exception as e:
|
|
self.logger.log(f"get_auth_user_info 에러: {e}", level=logging.ERROR, exc_info=True)
|
|
return {}
|
|
|
|
def get_full_user_info(self, user_id: str) -> dict:
|
|
self.logger.log(f"Entering get_full_user_info(user_id={user_id})", level=logging.DEBUG)
|
|
try:
|
|
user_resp = self.client.table("users").select("*").eq("id", user_id).execute()
|
|
if not user_resp.data:
|
|
self.logger.log("사용자 정보가 존재하지 않습니다.", level=logging.WARNING)
|
|
return {}
|
|
user_info = user_resp.data[0]
|
|
membership_level = user_info.get("membership_level", "default")
|
|
membership_resp = self.client.table("membership_levels").select("*").eq("level", membership_level).execute()
|
|
membership_info = membership_resp.data[0] if membership_resp.data else {}
|
|
full_info = {**user_info, "membership_level_data": membership_info}
|
|
self.logger.log(f"Full user info retrieved: {full_info}", level=logging.DEBUG)
|
|
return full_info
|
|
except Exception as e:
|
|
self.logger.log(f"get_full_user_info 에러: {e}", level=logging.ERROR, exc_info=True)
|
|
return {}
|
|
|
|
def update_last_login(self, user_id: str):
|
|
self.logger.log(f"Entering update_last_login(user_id={user_id})", level=logging.DEBUG)
|
|
try:
|
|
now_iso = datetime.now(timezone.utc).isoformat()
|
|
self.client.table("users").update({"last_login": now_iso}).eq("id", user_id).execute()
|
|
self.logger.log(f"Last login updated for user {user_id}", level=logging.INFO)
|
|
except Exception as e:
|
|
self.logger.log(f"update_last_login 에러: {e}", level=logging.ERROR, exc_info=True)
|
|
self.logger.log("Exiting update_last_login()", level=logging.DEBUG)
|