Spaces:
Running
Running
File size: 2,627 Bytes
565a379 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 |
import firebase_admin
from firebase_admin import credentials, firestore
import logging
from typing import Dict, Any, Optional, List
from app.core.settings import settings
logger = logging.getLogger(__name__)
class FirestoreClient:
"""
Wrapper for Firebase Firestore.
Handles real-time data storage and session management.
"""
def __init__(self):
self.db = None
self._initialize()
def _initialize(self):
"""
Initialize Firebase Admin SDK if not already initialized.
"""
try:
# Check if already initialized
if not firebase_admin._apps:
# In production, use default credentials (GOOGLE_APPLICATION_CREDENTIALS)
# Or use service account path from settings
if settings.FIREBASE_CREDENTIALS_PATH:
cred = credentials.Certificate(settings.FIREBASE_CREDENTIALS_PATH)
firebase_admin.initialize_app(cred)
else:
# Use Cloud Run / Default credentials
firebase_admin.initialize_app()
self.db = firestore.client()
logger.info("Firestore initialized successfully.")
except Exception as e:
logger.error(f"Failed to initialize Firestore: {e}")
self.db = None
def save_session(self, user_id: str, session_data: Dict[str, Any]):
"""
Save user session data in real-time.
"""
if not self.db:
return
try:
doc_ref = self.db.collection('sessions').document(user_id)
doc_ref.set(session_data, merge=True)
except Exception as e:
logger.error(f"Error saving session to Firestore: {e}")
def get_session(self, user_id: str) -> Optional[Dict[str, Any]]:
"""
Retrieve user session.
"""
if not self.db:
return None
try:
doc = self.db.collection('sessions').document(user_id).get()
if doc.exists:
return doc.to_dict()
return None
except Exception as e:
logger.error(f"Error getting session from Firestore: {e}")
return None
def add_solution(self, solution_data: Dict[str, Any]):
"""
Add a solved problem to the solutions collection.
"""
if not self.db:
return
try:
self.db.collection('solutions').add(solution_data)
except Exception as e:
logger.error(f"Error adding solution to Firestore: {e}")
|