| 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: |
| |
| 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}") |
