# A database storing dictionaries, keyed on a timestamp. # value = A dict which will be stored as a JSON object encoded in # UTF-8. Note that dict keys of type integer or float will become # strings while values will keep their type. # Note that there's a (slim) chance that you'd stomp on the previous # value if you're too quick with generating the timestamps, ie # invoking time.time() several times quickly enough. import time import couch import index class DictDB(): def __init__(self, database, hostname, username, password): self.server = couch.client.Server( f"http://{username}:{password}@{hostname}:5984/") try: self.couchdb = self.server.database(database) except couch.exceptions.NotFound: print("Creating database and indexes.") self.couchdb = self.server.create(database) self.server.create('_users') for i in index.indexes: self.couchdb.index(i) self._ts = time.time() def unique_key(self): ts = time.time() while ts == self._ts: ts = time.time() self._ts = round(ts * 1000) return self._ts def add(self, data, batch_write=False): key = str(self.unique_key()) if type(data) is list: for item in data: item['_id'] = key self.couchdb.save(item) else: data['_id'] = key self.couchdb.save(data) return key def get(self, key): return self.couchdb[key] def slice(self, key_from=None, key_to=None): pass def search(self, **kwargs): data = list() selector = dict() if kwargs: selector = {"selector": {}} for key in kwargs: if kwargs[key].isnumeric(): kwargs[key] = int(kwargs[key]) selector['selector'][key] = {'$eq': kwargs[key]} for doc in self.couchdb.find(selector): data.append(doc) return data