Spaces:
Runtime error
Runtime error
File size: 3,034 Bytes
ebd06cc |
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 |
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
|