oh, heres a second version that fixes list.__setitem__.
the operations here are not terribly efficient but you could build
better caching scenarios if needed.
from sqlalchemy import *
from sqlalchemy.orm import *
metadata = MetaData()
objects = Table('objects', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(50), nullable=False)
)
object_dict_of_lists = Table('object_dict_of_lists', metadata,
Column('id', Integer, primary_key=True),
Column('object_id', Integer, ForeignKey('objects.id')),
Column('key', String(50), nullable=False),
Column('data', String(50), nullable=False)
)
engine = create_engine("sqlite:///:memory:", echo=True)
metadata.bind = engine
metadata.create_all()
class Object(object):
def __init__(self, name, data=None):
self.name = name
if data:
self.data = data
class DictOfListElement(object):
def __init__(self, key, data):
self.key = key
self.data = data
mapper(Object, objects, properties={
'_data':relation(DictOfListElement)
})
mapper(DictOfListElement, object_dict_of_lists)
class ListAdapter(object):
def __init__(self, parent, key):
self.__parent = parent
self.__key = key
def __cached(self):
try:
return self.__cached_data
except AttributeError:
self.__cached_data = [item.data for item in
self.__parent._data if item.key == self.__key]
return self.__cached_data
__cached = property(__cached)
def __iter__(self):
return iter(self.__cached)
def __eq__(self, other):
return list(self) == list(other)
def __repr__(self):
return repr(list(self))
def append(self, item):
self.__parent._data.append(DictOfListElement(self.__key, item))
self.__cached.append(item)
def __getitem__(self, index):
return self.__cached[index]
def __setitem__(self, index, value):
[item for item in self.__parent._data if item.key ==
self.__key][index].data = value
self.__cached[index] = value
# other list like methods
class DictAdapterAttribute(object):
def __get__(self, instance, owner):
if instance is None:
return self
class DictAdapter(object):
def __getitem__(self, key):
return ListAdapter(instance, key)
def keys(self):
return iter(set([item.key for item in instance._data]))
def __eq__(self, other):
return dict(self) == dict(other)
def __repr__(self):
return repr(dict(self))
# other dict like methods
return DictAdapter()
def __set__(self, instance, somedict):
l =[]
for key, somelist in somedict.items():
for item in somelist:
l.append(DictOfListElement(key, item))
instance._data = l
Object.data = DictAdapterAttribute()
sess = create_session()
obj1 = Object('object1', {
'key1':['key1_one', 'key1_two', 'key1_three'],
'key2':['key2_one', 'key2_two', 'key2_three'],
'key3':['key3_one', 'key3_two', 'key3_three'],
})
sess.save(obj1)
sess.flush()
sess.clear()
obj1 = sess.query(Object).get(obj1.id)
assert obj1.data['key2'] == ['key2_one', 'key2_two', 'key2_three'],
obj1.data['key2']
obj1.data['key3'].append('key3_four')
obj1.data['key2'][1] = 'key2_two_modified'
sess.flush()
sess.clear()
obj1 = sess.query(Object).get(obj1.id)
assert obj1.data == {
'key1':['key1_one', 'key1_two', 'key1_three'],
'key2':['key2_one', 'key2_two_modified', 'key2_three'],
'key3':['key3_one', 'key3_two', 'key3_three', 'key3_four'],
}
print obj1.data
--~--~---------~--~----~------------~-------~--~----~
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to [email protected]
To unsubscribe from this group, send email to [EMAIL PROTECTED]
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en
-~----------~----~----~----~------~----~------~--~---
