| 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 |