from firebase_admin import credentials, db, initialize_app, auth import os import json from typing import Optional, Any, Dict class FirebaseClient: _instance = None def __new__(cls): if cls._instance is None: cls._instance = super(FirebaseClient, cls).__new__(cls) cls._instance._initialized = False return cls._instance def __init__(self): if self._initialized: return try: # Initialize Firebase credentials if os.path.exists('firebase-credentials.json'): cred = credentials.Certificate('firebase-credentials.json') else: cred_dict = json.loads(os.environ.get('FIRE_CRED')) cred = credentials.Certificate(cred_dict) self.app = initialize_app(cred, { 'databaseURL': os.environ.get('FIRE_URL') }) self._initialized = True except Exception as e: raise Exception(f"Firebase initialization failed: {e}") def get(self, path: str) -> Optional[Any]: """Read data from specified path""" try: ref = db.reference(path) return ref.get() except Exception as e: raise Exception(f"Firebase read error: {e}") def query_by_child(self, path: str, child_key: str, start_at: Optional[Any] = None, end_at: Optional[Any] = None) -> Optional[Dict]: """Query data ordered by child key with optional start/end filters""" try: ref = db.reference(path) query = ref.order_by_child(child_key) if start_at is not None: query = query.start_at(start_at) if end_at is not None: query = query.end_at(end_at) return query.get() except Exception as e: raise Exception(f"Firebase query error: {e}") def set(self, path: str, data: Dict) -> Dict: """Set data at specified path""" try: ref = db.reference(path) ref.set(data) return data except Exception as e: raise Exception(f"Firebase write error: {e}") def update(self, path: str, data: Dict) -> Dict: """Update data at specified path""" try: ref = db.reference(path) ref.update(data) return data except Exception as e: raise Exception(f"Firebase update error: {e}") def delete(self, path: str) -> bool: """Delete data at specified path""" try: ref = db.reference(path) ref.delete() return True except Exception as e: raise Exception(f"Firebase delete error: {e}") def push(self, path: str, data: Dict) -> str: """Push new data to specified path, returns generated key""" try: ref = db.reference(path) new_ref = ref.push(data) return new_ref.key except Exception as e: raise Exception(f"Firebase push error: {e}") def get_user_details(self, user_id: str) -> Dict: """Get authentication details for a user""" try: user_record = auth.get_user(user_id) return user_record except Exception as e: raise Exception(f"Firebase error fetching user data: {e}") def delete_user(self, user_id: str) -> Optional[Any]: """Delete a user""" try: auth.delete_user(user_id) except Exception as e: raise Exception(f"Firebase error fetching user data: {e}")