memory (persistent, sqlite): better private values handling + new LazyPersistentBinaryDict:
- merged private values method handling in sqlite, and added a keys arguments to get only some keys
- LazyPersistentBinaryDict allow to get values in database only when they are needed, saving memory for big data
This commit is contained in:
Goffi
parent
25c07ce105
commit
ceab0c6bdc
2 changed files with 194 additions and 187 deletions
|
|
@ -28,30 +28,38 @@ class MemoryNotInitializedError(Exception):
|
|||
|
||||
class PersistentDict(object):
|
||||
r"""A dictionary which save persistently each value assigned
|
||||
|
||||
/!\ be careful, each assignment means a database write
|
||||
/!\ Memory must be initialised before loading/setting value with instances of this class"""
|
||||
storage = None
|
||||
binary = False
|
||||
|
||||
def __init__(self, namespace, profile=None):
|
||||
"""@param namespace: unique namespace for this dictionary
|
||||
@param profile: profile which *MUST* exists, or None for general values"""
|
||||
"""
|
||||
|
||||
@param namespace: unique namespace for this dictionary
|
||||
@param profile(unicode, None): profile which *MUST* exists, or None for general values
|
||||
"""
|
||||
if not self.storage:
|
||||
log.error(_("PersistentDict can't be used before memory initialisation"))
|
||||
raise MemoryNotInitializedError
|
||||
self._cache = {}
|
||||
self._cache = None
|
||||
self.namespace = namespace
|
||||
self.profile = profile
|
||||
|
||||
def _setCache(self, data):
|
||||
self._cache = data
|
||||
|
||||
def load(self):
|
||||
"""Load persistent data from storage.
|
||||
|
||||
need to be called before any other operation
|
||||
@return: defers the PersistentDict instance itself
|
||||
"""
|
||||
if not self.profile:
|
||||
d = self.storage.loadGenPrivates(self._cache, self.namespace)
|
||||
else:
|
||||
d = self.storage.loadIndPrivates(self._cache, self.namespace, self.profile)
|
||||
return d.addCallback(lambda dummy: self)
|
||||
d = self.storage.getPrivates(self.namespace, binary=self.binary, profile=self.profile)
|
||||
d.addCallback(self._setCache)
|
||||
d.addCallback(lambda dummy: self)
|
||||
return d
|
||||
|
||||
def iteritems(self):
|
||||
return self._cache.iteritems()
|
||||
|
|
@ -102,17 +110,11 @@ class PersistentDict(object):
|
|||
return self._cache.__getitem__(key)
|
||||
|
||||
def __setitem__(self, key, value):
|
||||
if not self.profile:
|
||||
self.storage.setGenPrivate(self.namespace, key, value)
|
||||
else:
|
||||
self.storage.setIndPrivate(self.namespace, key, value, self.profile)
|
||||
self.storage.setPrivateValue(self.namespace, key, value, self.binary, self.profile)
|
||||
return self._cache.__setitem__(key, value)
|
||||
|
||||
def __delitem__(self, key):
|
||||
if not self.profile:
|
||||
self.storage.delGenPrivate(self.namespace, key)
|
||||
else:
|
||||
self.storage.delIndPrivate(self.namespace, key, self.profile)
|
||||
self.storage.detPrivateValue(self.namepace, key, self.binary, self.profile)
|
||||
return self._cache.__delitem__(key)
|
||||
|
||||
def get(self, key, default=None):
|
||||
|
|
@ -127,40 +129,103 @@ class PersistentDict(object):
|
|||
|
||||
def force(self, name):
|
||||
"""Force saving of an attribute to storage
|
||||
@return: deferred fired when data is actually saved"""
|
||||
if not self.profile:
|
||||
return self.storage.setGenPrivate(self.namespace, name, self._cache[name])
|
||||
return self.storage.setIndPrivate(self.namespace, name, self._cache[name], self.profile)
|
||||
|
||||
@return: deferred fired when data is actually saved
|
||||
"""
|
||||
return self.storage.setPrivateValue(self.namespace, name, self._cache[name], self.binary, self.profile)
|
||||
|
||||
|
||||
class PersistentBinaryDict(PersistentDict):
|
||||
"""Persistent dict where value can be any python data (instead of string only)"""
|
||||
binary = True
|
||||
|
||||
|
||||
class LazyPersistentBinaryDict(PersistentBinaryDict):
|
||||
ur"""PersistentBinaryDict which get key/value when needed
|
||||
|
||||
This Persistent need more database access, it is suitable for largest data,
|
||||
to save memory.
|
||||
/!\ most of methods return a Deferred
|
||||
"""
|
||||
# TODO: missing methods should be implemented using database access
|
||||
# TODO: a cache would be useful (which is deleted after a timeout)
|
||||
|
||||
def load(self):
|
||||
"""load persistent data from storage
|
||||
"""
|
||||
if not self.profile:
|
||||
return self.storage.loadGenPrivatesBinary(self._cache, self.namespace)
|
||||
else:
|
||||
return self.storage.loadIndPrivatesBinary(self._cache, self.namespace, self.profile)
|
||||
# we show a warning as calling load on LazyPersistentBinaryDict sounds like a code mistake
|
||||
log.warning(_(u"Calling load on LazyPersistentBinaryDict while it's not needed"))
|
||||
|
||||
def iteritems(self):
|
||||
raise NotImplementedError
|
||||
|
||||
def items(self):
|
||||
return self.storage.getPrivates(self.namespace, binary=self.binary, profile=self.profile)
|
||||
|
||||
def __repr__(self):
|
||||
raise NotImplementedError
|
||||
|
||||
def __str__(self):
|
||||
return "LazyPersistentBinaryDict (namespace: {})".format(self.namespace)
|
||||
|
||||
def __lt__(self, other):
|
||||
raise NotImplementedError
|
||||
|
||||
def __le__(self, other):
|
||||
raise NotImplementedError
|
||||
|
||||
def __eq__(self, other):
|
||||
raise NotImplementedError
|
||||
|
||||
def __ne__(self, other):
|
||||
raise NotImplementedError
|
||||
|
||||
def __gt__(self, other):
|
||||
raise NotImplementedError
|
||||
|
||||
def __ge__(self, other):
|
||||
raise NotImplementedError
|
||||
|
||||
def __cmp__(self, other):
|
||||
raise NotImplementedError
|
||||
|
||||
def __hash__(self):
|
||||
return hash(unicode(self.__class__) + self.namespace + (self.profile or u''))
|
||||
|
||||
def __nonzero__(self):
|
||||
raise NotImplementedError
|
||||
|
||||
def __contains__(self, key):
|
||||
raise NotImplementedError
|
||||
|
||||
def __iter__(self):
|
||||
raise NotImplementedError
|
||||
|
||||
def __getitem__(self, key):
|
||||
"""get the value as a Deferred"""
|
||||
return self.storage.getPrivates(self.namespace, keys=[key], binary=self.binary, profile=self.profile)
|
||||
|
||||
def __setitem__(self, key, value):
|
||||
if not self.profile:
|
||||
self.storage.setGenPrivateBinary(self.namespace, key, value)
|
||||
else:
|
||||
self.storage.setIndPrivateBinary(self.namespace, key, value, self.profile)
|
||||
return self._cache.__setitem__(key, value)
|
||||
self.storage.setPrivateValue(self.namespace, key, value, self.binary, self.profile)
|
||||
|
||||
def __delitem__(self, key):
|
||||
if not self.profile:
|
||||
self.storage.delGenPrivateBinary(self.namespace, key)
|
||||
else:
|
||||
self.storage.delIndPrivateBinary(self.namespace, key, self.profile)
|
||||
return self._cache.__delitem__(key)
|
||||
self.storage.detPrivateValue(self.namepace, key, self.binary, self.profile)
|
||||
|
||||
def force(self, name):
|
||||
def _valueOrDefault(self, value, default):
|
||||
if value is None:
|
||||
return default
|
||||
return value
|
||||
|
||||
def get(self, key, default=None):
|
||||
d = self.__getitem__(key)
|
||||
d.addCallback(self._valueOrDefault)
|
||||
return d
|
||||
|
||||
def setdefault(self, key, default):
|
||||
raise NotImplementedError
|
||||
|
||||
def force(self, name, value):
|
||||
"""Force saving of an attribute to storage
|
||||
@return: deferred fired when data is actually saved"""
|
||||
if not self.profile:
|
||||
return self.storage.setGenPrivateBinary(self.namespace, name, self._cache[name])
|
||||
return self.storage.setIndPrivateBinary(self.namespace, name, self._cache[name], self.profile)
|
||||
|
||||
@param value(object): value is needed for LazyPersistentBinaryDict
|
||||
@return: deferred fired when data is actually saved
|
||||
"""
|
||||
return self.storage.setPrivateValue(self.namespace, name, value, self.binary, self.profile)
|
||||
|
|
|
|||
|
|
@ -520,177 +520,119 @@ class SqliteStorage(object):
|
|||
return d
|
||||
|
||||
#Private values
|
||||
def loadGenPrivates(self, private_gen, namespace):
|
||||
"""Load general private values
|
||||
|
||||
@param private_gen: dictionary to fill
|
||||
@param namespace: namespace of the values
|
||||
@return: deferred
|
||||
def _privateDataEb(self, failure_, operation, namespace, key=None, profile=None):
|
||||
"""generic errback for data queries"""
|
||||
log.error(_(u"Can't {operation} data in database for namespace {namespace}{and_key}{for_profile}: {msg}").format(
|
||||
operation = operation,
|
||||
namespace = namespace,
|
||||
and_key = (u" and key " + key) if key is not None else u"",
|
||||
for_profile = (u' [' + profile + u']') if profile is not None else u'',
|
||||
msg = failure_))
|
||||
|
||||
def _generateDataDict(self, query_result, binary):
|
||||
if binary:
|
||||
return {k: pickle.loads(str(v)) for k,v in query_result}
|
||||
else:
|
||||
return dict(query_result)
|
||||
|
||||
def _getPrivateTable(self, binary, profile):
|
||||
"""Get table to use for private values"""
|
||||
table = [u'private']
|
||||
|
||||
if profile is None:
|
||||
table.append(u'gen')
|
||||
else:
|
||||
table.append(u'ind')
|
||||
|
||||
if binary:
|
||||
table.append(u'bin')
|
||||
|
||||
return u'_'.join(table)
|
||||
|
||||
def getPrivates(self, namespace, keys=None, binary=False, profile=None):
|
||||
"""Get private value(s) from databases
|
||||
|
||||
@param namespace(unicode): namespace of the values
|
||||
@param keys(iterable, None): keys of the values to get
|
||||
None to get all keys/values
|
||||
@param binary(bool): True to deserialise binary values
|
||||
@param profile(unicode, None): profile to use for individual values
|
||||
None to use general values
|
||||
@return (dict[unicode, object]): gotten keys/values
|
||||
"""
|
||||
log.debug(_(u"getting {type}{binary} private values from database for namespace {namespace}{keys}".format(
|
||||
type = u"general" if profile is None else "individual",
|
||||
binary = u" binary" if binary else u"",
|
||||
namespace = namespace,
|
||||
keys = u" with keys {}".format(u", ".join(keys)) if keys is not None else u"")))
|
||||
table = self._getPrivateTable(binary, profile)
|
||||
query_parts = [u"SELECT key,value FROM", table, "WHERE namespace=?"]
|
||||
args = [namespace]
|
||||
|
||||
def fillPrivates(result):
|
||||
for private in result:
|
||||
key, value = private
|
||||
private_gen[key] = value
|
||||
log.debug(_(u"loading general private values [namespace: %s] from database") % (namespace,))
|
||||
d = self.dbpool.runQuery("SELECT key,value FROM private_gen WHERE namespace=?", (namespace, )).addCallback(fillPrivates)
|
||||
return d.addErrback(lambda x: log.debug(_(u"No data present in database for namespace %s") % namespace))
|
||||
if keys is not None:
|
||||
query_parts.append(u'AND key IN ?')
|
||||
args.append(keys)
|
||||
|
||||
def loadIndPrivates(self, private_ind, namespace, profile):
|
||||
"""Load individual private values
|
||||
if profile is not None:
|
||||
query_parts.append(u'AND profile_id=?')
|
||||
args.append(self.profiles[profile])
|
||||
|
||||
@param privates_ind: dictionary to fill
|
||||
@param namespace: namespace of the values
|
||||
@param profile: a profile which *must* exist
|
||||
@return: deferred
|
||||
d = self.dbpool.runQuery(u" ".join(query_parts), args)
|
||||
d.addCallback(self._generateDataDict, binary)
|
||||
d.addErrback(self._privateDataEb, u"get", namespace, profile=profile)
|
||||
return d
|
||||
|
||||
def setPrivateValue(self, namespace, key, value, binary=False, profile=None):
|
||||
"""Set a private value in database
|
||||
|
||||
@param namespace(unicode): namespace of the values
|
||||
@param key(unicode): key of the value to set
|
||||
@param value(object): value to set
|
||||
@param binary(bool): True if it's a binary values
|
||||
binary values need to be serialised, used for everything but strings
|
||||
@param profile(unicode, None): profile to use for individual value
|
||||
if None, it's a general value
|
||||
"""
|
||||
table = self._getPrivateTable(binary, profile)
|
||||
query_values_names = [u'namespace', u'key', u'value']
|
||||
query_values = [namespace, key]
|
||||
|
||||
def fillPrivates(result):
|
||||
for private in result:
|
||||
key, value = private
|
||||
private_ind[key] = value
|
||||
log.debug(_(u"loading individual private values [namespace: %s] from database") % (namespace,))
|
||||
d = self.dbpool.runQuery("SELECT key,value FROM private_ind WHERE namespace=? AND profile_id=?", (namespace, self.profiles[profile]))
|
||||
d.addCallback(fillPrivates)
|
||||
return d.addErrback(lambda x: log.debug(_(u"No data present in database for namespace %s") % namespace))
|
||||
if binary:
|
||||
value = sqlite3.Binary(pickle.dumps(value, 0))
|
||||
|
||||
def setGenPrivate(self, namespace, key, value):
|
||||
"""Save the general private value in database
|
||||
query_values.append(value)
|
||||
|
||||
if profile is not None:
|
||||
query_values_names.append(u'profile_id')
|
||||
query_values.append(self.profiles[profile])
|
||||
|
||||
query_parts = [u"REPLACE INTO", table, u'(', u','.join(query_values_names), u')',
|
||||
u"VALUES (", u",".join(u'?'*len(query_values_names)), u')']
|
||||
|
||||
d = self.dbpool.runQuery(u" ".join(query_parts), query_values)
|
||||
d.addErrback(self._privateDataEb, u"set", namespace, key, profile=profile)
|
||||
return d
|
||||
|
||||
def delPrivateValue(self, namespace, key, binary=False, profile=None):
|
||||
"""Delete private value from database
|
||||
|
||||
@param category: category of the privateeter
|
||||
@param key: key of the private value
|
||||
@param value: value to set
|
||||
@return: deferred
|
||||
@param binary(bool): True if it's a binary values
|
||||
@param profile(unicode, None): profile to use for individual value
|
||||
if None, it's a general value
|
||||
"""
|
||||
d = self.dbpool.runQuery("REPLACE INTO private_gen(namespace,key,value) VALUES (?,?,?)", (namespace, key, value))
|
||||
d.addErrback(lambda ignore: log.error(_(u"Can't set general private value (%(key)s) [namespace:%(namespace)s] in database" %
|
||||
{"namespace": namespace, "key": key})))
|
||||
table = self._getPrivateTable(binary, profile)
|
||||
query_parts = [u"DELETE FROM", table, u"WHERE namespace=? AND key=?"]
|
||||
args = [namespace, key]
|
||||
if profile is not None:
|
||||
query_parts.append(u"AND profile_id=?")
|
||||
args.append(self.profiles[profile])
|
||||
d = self.dbpool.runQuery(u" ".join(query_parts), args)
|
||||
d.addErrback(self._privateDataEb, u"delete", namespace, key, profile=profile)
|
||||
return d
|
||||
|
||||
def setIndPrivate(self, namespace, key, value, profile):
|
||||
"""Save the individual private value in database
|
||||
|
||||
@param namespace: namespace of the value
|
||||
@param key: key of the private value
|
||||
@param value: value to set
|
||||
@param profile: a profile which *must* exist
|
||||
@return: deferred
|
||||
"""
|
||||
d = self.dbpool.runQuery("REPLACE INTO private_ind(namespace,key,profile_id,value) VALUES (?,?,?,?)", (namespace, key, self.profiles[profile], value))
|
||||
d.addErrback(lambda ignore: log.error(_(u"Can't set individual private value (%(key)s) [namespace: %(namespace)s] for [%(profile)s] in database" %
|
||||
{"namespace": namespace, "key": key, "profile": profile})))
|
||||
return d
|
||||
|
||||
def delGenPrivate(self, namespace, key):
|
||||
"""Delete the general private value from database
|
||||
|
||||
@param category: category of the privateeter
|
||||
@param key: key of the private value
|
||||
@return: deferred
|
||||
"""
|
||||
d = self.dbpool.runQuery("DELETE FROM private_gen WHERE namespace=? AND key=?", (namespace, key))
|
||||
d.addErrback(lambda ignore: log.error(_(u"Can't delete general private value (%(key)s) [namespace:%(namespace)s] in database" %
|
||||
{"namespace": namespace, "key": key})))
|
||||
return d
|
||||
|
||||
def delIndPrivate(self, namespace, key, profile):
|
||||
"""Delete the individual private value from database
|
||||
|
||||
@param namespace: namespace of the value
|
||||
@param key: key of the private value
|
||||
@param profile: a profile which *must* exist
|
||||
@return: deferred
|
||||
"""
|
||||
d = self.dbpool.runQuery("DELETE FROM private_ind WHERE namespace=? AND key=? AND profile=?)", (namespace, key, self.profiles[profile]))
|
||||
d.addErrback(lambda ignore: log.error(_(u"Can't delete individual private value (%(key)s) [namespace: %(namespace)s] for [%(profile)s] in database" %
|
||||
{"namespace": namespace, "key": key, "profile": profile})))
|
||||
return d
|
||||
|
||||
def loadGenPrivatesBinary(self, private_gen, namespace):
|
||||
"""Load general private binary values
|
||||
|
||||
@param private_gen: dictionary to fill
|
||||
@param namespace: namespace of the values
|
||||
@return: deferred
|
||||
"""
|
||||
|
||||
def fillPrivates(result):
|
||||
for private in result:
|
||||
key, value = private
|
||||
private_gen[key] = pickle.loads(str(value))
|
||||
log.debug(_(u"loading general private binary values [namespace: %s] from database") % (namespace,))
|
||||
d = self.dbpool.runQuery("SELECT key,value FROM private_gen_bin WHERE namespace=?", (namespace, )).addCallback(fillPrivates)
|
||||
return d.addErrback(lambda x: log.debug(_(u"No binary data present in database for namespace %s") % namespace))
|
||||
|
||||
def loadIndPrivatesBinary(self, private_ind, namespace, profile):
|
||||
"""Load individual private binary values
|
||||
|
||||
@param privates_ind: dictionary to fill
|
||||
@param namespace: namespace of the values
|
||||
@param profile: a profile which *must* exist
|
||||
@return: deferred
|
||||
"""
|
||||
|
||||
def fillPrivates(result):
|
||||
for private in result:
|
||||
key, value = private
|
||||
private_ind[key] = pickle.loads(str(value))
|
||||
log.debug(_(u"loading individual private binary values [namespace: %s] from database") % (namespace,))
|
||||
d = self.dbpool.runQuery("SELECT key,value FROM private_ind_bin WHERE namespace=? AND profile_id=?", (namespace, self.profiles[profile]))
|
||||
d.addCallback(fillPrivates)
|
||||
return d.addErrback(lambda x: log.debug(_(u"No binary data present in database for namespace %s") % namespace))
|
||||
|
||||
def setGenPrivateBinary(self, namespace, key, value):
|
||||
"""Save the general private binary value in database
|
||||
|
||||
@param category: category of the privateeter
|
||||
@param key: key of the private value
|
||||
@param value: value to set
|
||||
@return: deferred
|
||||
"""
|
||||
d = self.dbpool.runQuery("REPLACE INTO private_gen_bin(namespace,key,value) VALUES (?,?,?)", (namespace, key, sqlite3.Binary(pickle.dumps(value, 0))))
|
||||
d.addErrback(lambda ignore: log.error(_(u"Can't set general private binary value (%(key)s) [namespace:%(namespace)s] in database" %
|
||||
{"namespace": namespace, "key": key})))
|
||||
return d
|
||||
|
||||
def setIndPrivateBinary(self, namespace, key, value, profile):
|
||||
"""Save the individual private binary value in database
|
||||
|
||||
@param namespace: namespace of the value
|
||||
@param key: key of the private value
|
||||
@param value: value to set
|
||||
@param profile: a profile which *must* exist
|
||||
@return: deferred
|
||||
"""
|
||||
d = self.dbpool.runQuery("REPLACE INTO private_ind_bin(namespace,key,profile_id,value) VALUES (?,?,?,?)", (namespace, key, self.profiles[profile], sqlite3.Binary(pickle.dumps(value, 0))))
|
||||
d.addErrback(lambda ignore: log.error(_(u"Can't set individual binary private value (%(key)s) [namespace: %(namespace)s] for [%(profile)s] in database" %
|
||||
{"namespace": namespace, "key": key, "profile": profile})))
|
||||
return d
|
||||
|
||||
def delGenPrivateBinary(self, namespace, key):
|
||||
"""Delete the general private binary value from database
|
||||
|
||||
@param category: category of the privateeter
|
||||
@param key: key of the private value
|
||||
@return: deferred
|
||||
"""
|
||||
d = self.dbpool.runQuery("DELETE FROM private_gen_bin WHERE namespace=? AND key=?", (namespace, key))
|
||||
d.addErrback(lambda ignore: log.error(_(u"Can't delete general private binary value (%(key)s) [namespace:%(namespace)s] in database" %
|
||||
{"namespace": namespace, "key": key})))
|
||||
return d
|
||||
|
||||
def delIndPrivateBinary(self, namespace, key, profile):
|
||||
"""Delete the individual private binary value from database
|
||||
|
||||
@param namespace: namespace of the value
|
||||
@param key: key of the private value
|
||||
@param profile: a profile which *must* exist
|
||||
@return: deferred
|
||||
"""
|
||||
d = self.dbpool.runQuery("DELETE FROM private_ind_bin WHERE namespace=? AND key=? AND profile=?)", (namespace, key, self.profiles[profile]))
|
||||
d.addErrback(lambda ignore: log.error(_(u"Can't delete individual private binary value (%(key)s) [namespace: %(namespace)s] for [%(profile)s] in database" %
|
||||
{"namespace": namespace, "key": key, "profile": profile})))
|
||||
return d
|
||||
##Helper methods##
|
||||
|
||||
def __getFirstResult(self, result):
|
||||
|
|
|
|||
Loading…
Reference in a new issue