sugarlabs/sugar-toolkit-gtk3
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

datastore fixes

Browse Source
This commit is contained in:
Dan Williams 2006-12-11 17:25:48 -05:00
parent dbca112368
commit 039ac19432
4 changed files with 170 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
services/datastore/datastore.py
View File

@ -72,8 +72,7 @@ class DataStoreDBusHelper(dbus.service.Object):
@dbus.service.method(_DS_DBUS_INTERFACE,
in_signature="x", out_signature="o")
def get(self, uid):
uid = self._parent.get(uid)
return self._create_op(uid)
return _create_op(self._parent.get(uid))
@dbus.service.method(_DS_DBUS_INTERFACE,
in_signature="aya{sv}", out_signature="o")
@ -205,7 +204,8 @@ class DataStore(object):
def create(self, data, prop_dict=None):
curs = self._dbcx.cursor()
curs.execute("INSERT INTO objects (uid) VALUES (NULL);")
data = sqlite.encode(_get_data_as_string(data))
curs.execute("INSERT INTO objects (uid, data) VALUES (NULL, '%s');" % data)
curs.execute("SELECT last_insert_rowid();")
rows = curs.fetchall()
self._dbcx.commit()
@ -213,8 +213,8 @@ class DataStore(object):
uid = last_row[0]
for (key, value) in prop_dict.items():
safe_key = key.replace("'", "''")
value = str(value)
curs.execute("INSERT INTO properties (objid, key, value) VALUES (%d, '%s', '%s');" % (uid, safe_key, sqlite.encode(value)))
value = sqlite.encode(_get_data_as_string(value))
curs.execute("INSERT INTO properties (objid, key, value) VALUES (%d, '%s', '%s');" % (uid, safe_key, value))
self._dbcx.commit()
del curs
return uid
@ -229,18 +229,20 @@ class DataStore(object):
return 0
def find(self, prop_dict):
query = "SELECT objid FROM properties WHERE ("
query = "SELECT objid FROM properties"
subquery = ""
for (key, value) in prop_dict.items():
safe_key = key.replace("'", "''")
value = _get_data_as_string(value)
if not len(value):
raise ValueError("Property values must not be blank.")
value = str(value)
substr = "key='%s' AND value='%s'" % (safe_key, sqlite.encode(value))
substr = "(key='%s' AND value='%s')" % (safe_key, sqlite.encode(value))
if len(subquery) > 0:
subquery += " AND"
subquery += " OR "
subquery += substr
query += subquery + ")"
if len(subquery):
query += " WHERE (%s)" % subquery
query += ";"
curs = self._dbcx.cursor()
curs.execute(query)
rows = curs.fetchall()
@ -260,8 +262,8 @@ class DataStore(object):
if len(res) <= 0:
del curs
raise NotFoundError("Object %d was not found." % uid)
data = _get_data_as_string(data)
curs.execute("UPDATE objects SET data='%s' WHERE uid=%d;" % (sqlite.encode(data), uid))
data = sqlite.encode(_get_data_as_string(data))
curs.execute("UPDATE objects SET data='%s' WHERE uid=%d;" % (data, uid))
self._dbcx.commit()
del curs
self._dbus_obj_helper.Updated(True, {}, False, uid=uid)
@ -282,11 +284,12 @@ class DataStore(object):
for (key, value) in prop_dict.items():
safe_key = key.replace("'", "''")
value = _get_data_as_string(value)
if not len(value):
# delete the property
curs.execute("DELETE FROM properties WHERE (objid=%d AND key='%s');" % (uid, safe_key))
else:
enc_value = sqlite.encode(_get_data_as_string(value))
enc_value = sqlite.encode(value)
curs.execute("SELECT objid FROM properties WHERE (objid=%d AND key='%s');" % (uid, safe_key))
if len(curs.fetchall()) > 0:
curs.execute("UPDATE properties SET value='%s' WHERE (objid=%d AND key='%s');" % (enc_value, uid, safe_key))
@ -301,20 +304,23 @@ class DataStore(object):
curs.execute('SELECT uid, data FROM objects WHERE uid=%d;' % uid)
res = curs.fetchall()
self._dbcx.commit()
del curs
if len(res) <= 0:
raise NotFoundError("Object %d was not found." % uid)
return sqlite.decode(res[0]['data'])
data = res[0][1]
del curs
return data
def get_properties(self, uid, keys):
query = "SELECT objid, key, value FROM properties WHERE (objid=%d AND (" % uid
query = "SELECT objid, key, value FROM properties WHERE (objid=%d" % uid
subquery = ""
if len(keys) > 0:
for key in keys:
if len(subquery) > 0:
subquery += " OR "
safe_key = key.replace("'", "''")
subquery += "key='%s'" % safe_key
query += subquery + "));"
subquery += ")"
query += subquery + ");"
curs = self._dbcx.cursor()
curs.execute(query)
rows = curs.fetchall()
@ -322,7 +328,8 @@ class DataStore(object):
prop_dict = {}
for row in rows:
conv_key = row['key'].replace("''", "'")
prop_dict[conv_key] = row['value']
prop_dict[conv_key] = sqlite.decode(row['value'])
prop_dict['uid'] = str(uid)
del curs
return prop_dict

4
services/datastore/sugar-data-store Normal file → Executable file
View File

@ -28,5 +28,9 @@ sys.path.insert(0, env.get_services_dir())
logger.start('data-store')
logging.info('Starting the data store...')
try:
from datastore import datastore
except ImportError:
import datastore
datastore.main()

30
sugar/datastore/datastore.py
View File

@ -32,8 +32,10 @@ class ObjectCache(object):
self._cache[op] = obj
def remove(self, object_path):
if self._cache.has_key(object_path):
try:
del self._cache[object_path]
except IndexError:
pass
DS_DBUS_SERVICE = "org.laptop.sugar.DataStore"
@ -55,16 +57,21 @@ class DSObject(gobject.GObject):
self._ps_new_object = new_obj_cb
self._ps_del_object = del_obj_cb
bobj = bus.get_object(DS_DBUS_SERVICE, object_path)
self._dsobj = dbus.Interface(bobj, self.__DS_OBJECT_DBUS_INTERFACE)
self._dsobj = dbus.Interface(bobj, self._DS_OBJECT_DBUS_INTERFACE)
self._dsobj.connect_to_signal('Updated', self._updated_cb)
self._data = None
self._data_needs_update = True
self._properties = self._dsobj.get_properties()
self._properties = None
self._deleted = False
def object_path(self):
return self._object_path
def uid(self):
if not self._properties:
self._properties = self._dsobj.get_properties([])
return self._properties['uid']
def _emit_updated_signal(self, data, prop_dict, deleted):
self.emit('updated', data, prop_dict, deleted)
return False
@ -93,7 +100,10 @@ class DSObject(gobject.GObject):
def get_data(self):
if self._data_needs_update:
self._data = self._dsobj.get_data()
data = self._dsobj.get_data()
self._data = ""
for c in data:
self._data += chr(c)
return self._data
def set_data(self, data):
@ -116,12 +126,14 @@ class DSObject(gobject.GObject):
self._properties = old_props
raise e
def get_properties(self, prop_dict):
def get_properties(self, prop_list=[]):
if not self._properties:
self._properties = self._dsobj.get_properties(prop_list)
return self._properties
class DataStore(gobject.GObject):
_DS_DBUS_OBJECT_PATH = DBUS_PATH + "/Object/"
_DS_DBUS_OBJECT_PATH = DS_DBUS_PATH + "/Object/"
def __init__(self):
gobject.GObject.__init__(self)
@ -132,7 +144,9 @@ class DataStore(gobject.GObject):
def _new_object(self, object_path):
obj = self._objcache.get(object_path)
if not obj:
if obj:
return obj
if object_path.startswith(self._DS_DBUS_OBJECT_PATH):
obj = DSObject(self._bus, self._new_object,
self._del_object, object_path)
@ -146,7 +160,7 @@ class DataStore(gobject.GObject):
pass
def get(self, uid):
return self._new_object(self._ds.get(uid))
return self._new_object(self._ds.get(int(uid)))
def create(self, data, prop_dict={}):
op = self._ds.create(dbus.ByteArray(data), dbus.Dictionary(prop_dict))

107
tests/test-datastore.py Executable file
View File

@ -0,0 +1,107 @@
#!/usr/bin/python
# Copyright (C) 2006, Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
import unittest
from sugar.datastore import datastore
import dbus
class NotFoundError(dbus.DBusException): pass
_ds = datastore.get_instance()
class DataStoreTestCase(unittest.TestCase):
_TEST_DATA = "adsfkjadsfadskjasdkjf"
_TEST_PROPS = {'foo': 1, 'bar': 'baz'}
def _create_test_object(self):
obj = _ds.create(self._TEST_DATA, self._TEST_PROPS)
self.assert_(obj)
return obj
def testObjectCreate(self):
obj = self._create_test_object()
self.assert_(obj.uid())
_ds.delete(obj)
def testObjectGet(self):
# create a new object
obj = self._create_test_object()
self.assert_(obj.uid())
obj2 = _ds.get(obj.uid())
self.assert_(obj2)
_ds.delete(obj)
def testObjectDelete(self):
obj = self._create_test_object()
uid = obj.uid()
_ds.delete(obj)
try:
_ds.get(uid)
except dbus.DBusException, e:
if str(e).find("NotFoundError:") < 0:
self.fail("Expected a NotFoundError")
else:
self.fail("Expected a NotFoundError.")
def testObjectFind(self):
obj = self._create_test_object()
found = _ds.find(self._TEST_PROPS)
self.assert_(obj in found)
_ds.delete(obj)
def testObjectGetData(self):
obj = self._create_test_object()
data = obj.get_data()
self.assert_(data == self._TEST_DATA)
_ds.delete(obj)
_OTHER_DATA = "532532532532532;lkjkjkjfsakjfakjfdsakj"
def testObjectSetData(self):
obj = self._create_test_object()
data = obj.get_data()
self.assert_(data == self._TEST_DATA)
obj.set_data(self._OTHER_DATA)
data = obj.get_data()
self.assert_(data == self._OTHER_DATA)
_ds.delete(obj)
def testObjectGetProperties(self):
obj = self._create_test_object()
props = obj.get_properties()
for (key, value) in props.items():
if key == 'uid':
continue
self.assert_(key in self._TEST_PROPS)
self.assert_(str(self._TEST_PROPS[key]) == str(value))
for (key, value) in self._TEST_PROPS.items():
self.assert_(key in props)
self.assert_(str(props[key]) == str(value))
_ds.delete(obj)
def main():
dsTestSuite = unittest.TestSuite()
dsTestSuite.addTest(DataStoreTestCase('testObjectCreate'))
dsTestSuite.addTest(DataStoreTestCase('testObjectGet'))
dsTestSuite.addTest(DataStoreTestCase('testObjectDelete'))
dsTestSuite.addTest(DataStoreTestCase('testObjectFind'))
dsTestSuite.addTest(DataStoreTestCase('testObjectGetData'))
dsTestSuite.addTest(DataStoreTestCase('testObjectSetData'))
dsTestSuite.addTest(DataStoreTestCase('testObjectGetProperties'))
unittest.TextTestRunner(verbosity=2).run(dsTestSuite)
if __name__ == "__main__":
main()