File size: 3,410 Bytes
215f97f | 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 | """
Firebase Admin SDK initialization module.
Supports loading credentials from:
1. JSON key file path (FIREBASE_CREDENTIALS_PATH)
2. Raw JSON string from environment variable (FIREBASE_CREDENTIALS_JSON)
"""
import json
import os
from typing import Optional
import firebase_admin
from firebase_admin import credentials, firestore, storage
from app.core.config import settings
from app.core.logging import get_logger
logger = get_logger(__name__)
_firebase_app: Optional[firebase_admin.App] = None
def init_firebase() -> Optional[firebase_admin.App]:
"""Initialize Firebase Admin SDK app if credentials are provided."""
global _firebase_app
if _firebase_app is not None or firebase_admin._apps:
logger.info("Firebase Admin SDK already initialized.")
return firebase_admin.get_app()
cred = None
# Option 1: File path
if settings.FIREBASE_CREDENTIALS_PATH and os.path.exists(settings.FIREBASE_CREDENTIALS_PATH):
try:
cred = credentials.Certificate(settings.FIREBASE_CREDENTIALS_PATH)
logger.info("Loaded Firebase credentials from file: %s", settings.FIREBASE_CREDENTIALS_PATH)
except Exception as e:
logger.error("Failed to load Firebase credentials from file %s: %s", settings.FIREBASE_CREDENTIALS_PATH, str(e))
# Option 2: JSON string from ENV
elif settings.FIREBASE_CREDENTIALS_JSON:
try:
cert_dict = json.loads(settings.FIREBASE_CREDENTIALS_JSON)
cred = credentials.Certificate(cert_dict)
logger.info("Loaded Firebase credentials from environment JSON string.")
except Exception as e:
logger.error("Failed to parse Firebase credentials from env JSON string: %s", str(e))
options = {}
if settings.FIREBASE_STORAGE_BUCKET:
options["storageBucket"] = settings.FIREBASE_STORAGE_BUCKET
if cred:
try:
_firebase_app = firebase_admin.initialize_app(cred, options=options if options else None)
logger.info("Firebase Admin SDK successfully initialized.")
return _firebase_app
except Exception as e:
logger.error("Failed to initialize Firebase Admin SDK app: %s", str(e))
else:
logger.warning(
"Firebase credentials not found (checked path: '%s'). "
"Firebase Admin SDK skipped. Place key file at path or set FIREBASE_CREDENTIALS_JSON.",
settings.FIREBASE_CREDENTIALS_PATH,
)
return None
def get_firestore_db():
"""Return initialized Firebase Firestore client, or None if not initialized."""
if not firebase_admin._apps:
init_firebase()
if firebase_admin._apps:
try:
return firestore.client()
except Exception as e:
logger.error("Failed to access Firestore client: %s", str(e))
return None
def get_storage_bucket():
"""Return initialized Firebase Storage bucket, or None if not initialized."""
if not firebase_admin._apps:
init_firebase()
if firebase_admin._apps:
try:
bucket_name = settings.FIREBASE_STORAGE_BUCKET or None
return storage.bucket(name=bucket_name)
except Exception as e:
logger.error("Failed to access Storage bucket: %s", str(e))
return None
|