Spaces:
Runtime error
Runtime error
import os | |
import json | |
import random | |
import logging | |
import firebase_admin | |
from firebase_admin import credentials, firestore | |
public_creds_path = 'public_creds.json' | |
def load_credentials(): | |
try: | |
with open(public_creds_path) as f: | |
credentials_dict = json.load(f) | |
secret = { | |
'private_key_id': os.environ.get('private_key_id'), | |
'private_key': os.environ.get('private_key').replace(r'\n', '\n') | |
} | |
credentials_dict.update(secret) | |
return credentials_dict | |
except Exception as e: | |
logging.error(f'Error while loading credentials: {e}') | |
return None | |
def connect_to_db(credentials_dict): | |
try: | |
cred = credentials.Certificate(credentials_dict) | |
if not firebase_admin._apps: | |
firebase_admin.initialize_app(cred) | |
logging.info('Established connection to db!') | |
return firestore.client() | |
except Exception as e: | |
logging.error(f'Error while connecting to db: {e}') | |
return None | |
def get_statements_from_db(db): | |
try: | |
document = db.collection('ItemDesirability').document('Items') | |
statements = document.get().to_dict()['statements'] | |
logging.info(f'Retrieved {len(statements)} statements from db!') | |
return statements | |
except Exception as e: | |
logging.error(f'Error while retrieving items from db: {e}') | |
return None | |
def pick_random(input_list): | |
try: | |
return random.choice(input_list) | |
except Exception as e: | |
logging.error(f'Error while picking random statement: {e}') | |
return None | |
def write_to_db(db, payload): | |
try: | |
collection_ref = db.collection('ItemDesirability') | |
doc_ref = collection_ref.document('Responses') | |
doc = doc_ref.get() | |
if doc.exists: | |
doc_ref.update({ | |
'Data': firestore.ArrayUnion([payload]) | |
}) | |
else: | |
doc_ref.set({ | |
'Data': [payload] | |
}) | |
logging.info(f'Sent payload to db!') | |
return True | |
except Exception as e: | |
logging.error(f'Error while sending payload to db: {e}') | |
return False |