File size: 3,670 Bytes
544c0b2 bcdd77d 544c0b2 | 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 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 | 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}") |