Spaces:
Runtime error
Runtime error
| from firebase import firebase | |
| import os,json | |
| import baseInfra.firebase_handler as fbh | |
| secret=os.environ['FIREBASE_TOKEN'] | |
| user_id="131251" | |
| auth=firebase.FirebaseAuthentication(secret=secret,email="[email protected]",extra={"id":user_id}) | |
| fb_url="https://device-1a455.firebaseio.com" | |
| fb= firebase.FirebaseApplication(fb_url,auth) | |
| base_dir="/users/"+user_id | |
| class DbInterface: | |
| """ | |
| A database interface. | |
| This class provides a dummy implementation of a database interface. It has two methods: `get_task_list()` and `update_task_list()`. | |
| """ | |
| def __init__(self): | |
| self.base_dir=base_dir | |
| self.conf_loc="config" | |
| self.cache_loc="vs_cache" | |
| def _set_base_dir(self,sub_dir): | |
| """ | |
| This internal method is primarily for unit testcases | |
| Adds extra subdir to the path | |
| """ | |
| if sub_dir.startswith("/"): | |
| self.base_dir=base_dir+sub_dir | |
| else: | |
| self.base_dir=base_dir+"/"+sub_dir | |
| if self.base_dir.endswith("/"): | |
| self.base_dir=self.base_dir[:-1] | |
| return self.base_dir | |
| def get_config(self,path): | |
| """ | |
| Gets the config from the database given path | |
| This is used for getting configs of executor, planner or tools | |
| Returns: | |
| The json config object. | |
| """ | |
| print(self.base_dir+"/"+self.conf_loc) | |
| print(path) | |
| config = fb.get(self.base_dir+"/"+self.conf_loc, path) | |
| if config is None: | |
| return {} | |
| return config | |
| def get_matching_cache(self,input): | |
| """ | |
| Gets the matching cache from the database given input | |
| This is used for getting similar documents from the vector store | |
| Returns: | |
| The json cache object. | |
| """ | |
| fb_key=fbh.convert_to_firebase_key(input) | |
| cache = fb.get(self.base_dir+"/"+self.cache_loc, fb_key) | |
| if cache == None: | |
| return [] | |
| else: | |
| for item in cache: #ie cache is list of dicts {'input':inStr,'value':cached_items_list} | |
| if item['input'] == input: | |
| return item['value'] | |
| return [] | |
| def add_to_cache(self,input,value): | |
| """ | |
| Adds the input and value to the cache | |
| This is used for adding documents to the vector store | |
| Returns: | |
| The True if item in cache was updated and false if a new item was added | |
| """ | |
| retVal=False | |
| fb_key=fbh.convert_to_firebase_key(input) | |
| cache = fb.get(self.base_dir+"/"+self.cache_loc, fb_key) | |
| if cache is None: | |
| cache = [] | |
| else: | |
| for item in cache: #ie cache is list of dicts {'input':inStr,'value':cached_items_list} | |
| if item['input'] == input: | |
| item['value']=value | |
| retVal=True | |
| cache.append({'input':input,'value':value}) | |
| fb.patch(self.base_dir+"/"+self.cache_loc, {fb_key:cache}) | |
| return retVal | |