Hi all -- I am relatively new to sqlalchemy and am in the midst of building a generic "vertical" dictionary collection base class that can be subclassed and extended in different ways. I began by examining the "dictlike.py" and "dictlike-polymorphic" SQLalchemy examples and extended them into a class factory framework using metaclasses. This is my first attempt at using metaclasses and class-level programming in python. Although my implementation works, I feel that contributions and suggestions from all of you would substantially improve the design. Any comments and/or feedback would be appreciated -- and I apologize for the somewhat complicated code.
A summary of my approach is as follows: 1) Create a method to automatically generated the database tables for a new vertical parent class 2) Create a method to create the class types to use in mapping to the tables 3) Instrument and connect a new class to be "vertically enabled" using a metaclass method 4) Provide a simple interface that the new class can inherit via the VerticalAttrMixin class to provide methods for accessing the key/ values in the "vertical" dictionary collection. The code is provided below: ---------------------------------- from datetime import datetime from sqlalchemy import Table, Column, DateTime, String, Integer, Enum, Boolean, Float, Sequence, ForeignKey, UniqueConstraint, ForeignKeyConstraint from sqlalchemy.orm import mapper, relationship from sqlalchemy.orm.collections import attribute_mapped_collection, MappedCollection from sqlalchemy.ext.associationproxy import association_proxy from sqlalchemy.ext.declarative import DeclarativeMeta from sqlalchemy.orm.exc import MultipleResultsFound,NoResultFound from sequel2.db import metadata, Base, Session, SequelDBError ENUM_TYPE = "enum" _value_types = Enum(ENUM_TYPE, "bool", "int", "float", name="value_type_enum") def create_vertical_attr_tables(parent_tablename): keys_tablename = parent_tablename + '_attr_keys' keys_pk_seq = parent_tablename + '_key_id_seq' keys_table = Table(keys_tablename, metadata, Column('id', Integer, Sequence(keys_pk_seq), primary_key=True), Column('name', String(50), unique=True), Column('type', _value_types, default=ENUM_TYPE), Column('description', String(1000))) values_tablename = parent_tablename + '_attr_enum_values' values_pk_seq = parent_tablename + '_enum_value_id_seq' values_table = Table(values_tablename, metadata, Column('key_id', ForeignKey(keys_tablename + '.id'), primary_key=True), Column('value_id', Integer, Sequence(values_pk_seq), primary_key=True), Column('value', String(50), nullable=False), Column('description', String(1000)), UniqueConstraint('key_id', 'value', name="value_unique_constraint")) attrs_tablename = parent_tablename + '_attrs' attrs_fk = parent_tablename + '_id' attrs_table = Table(attrs_tablename, metadata, # Column('parent_id', ForeignKey(parent_tablename + '.id'), primary_key=True), Column(attrs_fk, ForeignKey(parent_tablename + '.id'), primary_key=True), Column('key_id', ForeignKey(keys_tablename + '.id'), primary_key=True), Column('enum_value_id', ForeignKey(values_tablename + '.value_id')), Column('bool_value', Boolean, default=False), Column('int_value', Integer, default=0), Column('float_value', Float, default=0), # add some auditing columns Column('updated_at', DateTime, default=datetime.now), Column('updated_by', String(50), default="anonymous")) return keys_table, values_table, attrs_table class VerticalAttrKey(object): def __init__(self, name, type, description=None): self.name = name self.type = type self.description = description def __repr__(self): return ("<%s(id='%s',name='%s',type='%s',description='%s')>" % (self.__class__.__name__, self.id, self.name, self.type, self.description)) def __getitem__(self, value): assert self.type == ENUM_TYPE return self.enum_values[value] def __contains__(self, value): assert self.type == ENUM_TYPE return value in self.enum_values # del an enum value from this key def __delitem__(self, value): assert self.type == ENUM_TYPE del self.enum_values[value] def get(self, value, default): return self.enum_values.get(value, None) def get_enums(self): assert self.type == ENUM_TYPE return self.enum_values.keys() def set_enum(self, value, description=None, newvalue=None): assert self.type == ENUM_TYPE if value in self.enum_values: # update description of old value oldobj = self.enum_values[value] if (newvalue is not None) and (newvalue != oldobj.value): oldobj.value = newvalue if (description is not None) and (description != oldobj.description): oldobj.description = description else: self.enum_values[value] = self.value_class(value, description) def _parse_value_tuple(self, value_tuple): if isinstance(value_tuple, basestring): # scalar parameter value = str(value_tuple) description = None elif len(value_tuple) >= 2: value, description = value_tuple[0:2] value = str(value) else: raise SequelDBError("Invalid value_tuple %s passed to set_value" % value_tuple) return value, description def set_enums(self, value_tuples): assert self.type == ENUM_TYPE for value_tuple in value_tuples: value, description = self._parse_value_tuple(value_tuple) self.set_enum(value=value, description=description) class VerticalAttrEnumValue(object): # sequentially assign ids (only needed for SQLite) current_id = 1 def __init__(self, value=None, description=None): # TODO: only for SQLite self.value_id = VerticalAttrEnumValue.current_id VerticalAttrEnumValue.current_id += 1 self.value = value self.description = description def __repr__(self): return ("<%s(key_id='%s',value='%s',description='%s')>" % (self.__class__.__name__, self.key_id, self.value, self.description)) class VerticalAttr(object): _key_field_map = {ENUM_TYPE: 'enum_value', 'bool': 'bool_value', 'int': 'int_value', 'float': 'float_value'} def __init__(self, parent=None, key=None, value=None, updated_by=None): self.parent = parent self.key = key self._set_value(value) if updated_by is not None: self.updated_by = updated_by def __repr__(self): parent_name = None if self.parent is None else self.parent.name key_name = None if self.key is None else self.key.name return ("<%s(parent='%s',key='%s',value='%s')>" % (self.__class__.__name__, parent_name, key_name, self.value)) def _set_value(self, value): key_type = self._key_type_map[self.key.type] #print 'setting attr', 'self', type(self), 'key', self._key_field_map[self.key.type], 'value', value #print 'value type', type(value) #rint 'key type', key_type #if not (isinstance(value, key_type) or isinstance(value, type(None))): if not (isinstance(value, key_type)): # try to coerce to correct type value = key_type(value) #raise TypeError(type(value)) setattr(self, self._key_field_map[self.key.type], value) def _get_value(self): field_name = self._key_field_map[self.key.type] return getattr(self, field_name) def _del_value(self): self._set_value(None) value = property(_get_value, _set_value, _del_value, doc="""get/set value based on type""") def create_vertical_attr_classes(parent_classname): key_classname = parent_classname + 'AttrKey' key_class = type(key_classname, (VerticalAttrKey,), {}) enum_value_classname = parent_classname + 'AttrEnumValue' enum_value_class = type(enum_value_classname, (VerticalAttrEnumValue,), {}) # need to customize the key_type_map for the Attribute class # so that enum keys map to their corresponding value types _key_type_map = {ENUM_TYPE: enum_value_class, 'bool': bool, 'int': int, 'float': float} attr_classname = parent_classname + 'Attr' attr_class = type(attr_classname, (VerticalAttr,), {'_key_type_map': _key_type_map}) # link key/value classes together setattr(key_class, 'value_class', enum_value_class) setattr(enum_value_class, 'key_class', key_class) return key_class, enum_value_class, attr_class def create_vertical_attr_objects(parent_classname, parent_tablename): keys_table, enum_values_table, attrs_table = create_vertical_attr_tables(parent_tablename) key_class, enum_value_class, attr_class = create_vertical_attr_classes(parent_classname) mapper(key_class, keys_table, properties = {'attrs': relationship(attr_class, backref='key', cascade="all, delete, delete-orphan"), 'enum_values': relationship(enum_value_class, backref='key', cascade="all, delete, delete-orphan", collection_class=attribute_mapped_collection('value'))}) mapper(enum_value_class, enum_values_table, properties = {'attrs': relationship(attr_class, backref='enum_value', cascade="all, delete, delete-orphan")}) mapper(attr_class, attrs_table) return keys_table, key_class, enum_values_table, enum_value_class, attrs_table, attr_class class VerticalAttrMetaclass(DeclarativeMeta): def __init__(cls, name, bases, dct): super(VerticalAttrMetaclass, cls).__init__(name, bases, dct) #print "Creating class %s using VerticalAttrMetaclass" % name cls.keys_table, cls.key_class, cls.enum_values_table, cls.enum_value_class, cls.attrs_table, cls.attr_class = create_vertical_attr_objects(name, cls.__tablename__) # map the class to its personal VerticalAttr class, # but hide this relationship since it is less user friendly cls._attrs = relationship(cls.attr_class, backref="parent", cascade="all, delete, delete- orphan", collection_class=attribute_mapped_collection('key')) # proxy the 'value' attribute from the 'attrs' relationship to # remove the need to deal with VerticalAttr objects entirely def _create_vertical_attr(key, value): """A creator function""" return cls.attr_class(key=key, value=value) cls.attrs = association_proxy('_attrs', 'value', creator=_create_vertical_attr) # create a convenient lookup by key name instead of key object cls._attrs_by_name = relationship(cls.attr_class, viewonly=True, collection_class=attribute_mapped_collection('key.name')) cls.attrs_by_name = association_proxy('_attrs_by_name', 'value', creator=None) # define interface functions def query_key(name): '''get attribute key if exists, None otherwise''' session = Session() try: return session.query(cls.key_class).filter_by(name=name).one() except NoResultFound: return None cls.query_key = staticmethod(query_key) def query_keys(): session = Session() return session.query(cls.key_class).all() cls.query_keys = staticmethod(query_keys) def set_key(key_name, key_type=None, description=None, values=None, new_key_name=None): key = cls.query_key(key_name) if key is None: # create it key = cls.key_class(key_name, key_type, description) else: # update name/description if new_key_name is not None: key.name = new_key_name if description is not None: key.description = description # update values if (values is not None) and (len(values) > 0): key.set_enums(values) return key cls.set_key = staticmethod(set_key) def create_key(name, type, description=None, values=None): k = cls.query_key(name) if k is not None: raise SequelDBError('Key %s already exists, use update_key to modify') k = cls.key_class(name, type, description) if (values is not None) and (len(values) > 0): k.set_enums(values) return k cls.create_key = staticmethod(create_key) def update_key(oldname, newname=None, description=None): session = Session() k = session.query(cls.key_class).filter_by(name=oldname).one() if newname is not None: k.name = newname if description is not None: k.description = description return k cls.update_key = staticmethod(update_key) def del_keys(names): session = Session() for name in names: k = session.query(cls.key_class).filter_by(name=name).one() session.delete(k) cls.del_keys = staticmethod(del_keys) class VerticalAttrMixin(object): def __init__(self): super(VerticalAttrMixin, self).__init__() def has_attr(self, name): if name in self.attrs_by_name: return True return False def get_attr(self, name): v = self.attrs_by_name[name] if isinstance(v, self.enum_value_class): return v.value return v def iter_attrs(self): for k,v in self.attrs.iteritems(): if k.type == ENUM_TYPE: yield k.name, v.value else: yield k.name, v def set_attr(self, key_name, value): keyobj = self.query_key(key_name) if keyobj is None: raise KeyError("Key '%s' not found" % key_name) if value is None: raise ValueError("Cannot set attribute %s to 'None'" % key_name) if (keyobj.type == ENUM_TYPE): value = keyobj[value] self.attrs[keyobj] = value def del_attr(self, key_name): key = self.query_key(key_name) if key not in self.attrs: raise KeyError("Key %s not found" % key_name) del self.attrs[key] def set_attrs(self, kv_pairs): for key_name, value in kv_pairs: self.set_attr(key_name, value) def del_attrs(self, names): for name in names: self.del_attr(name) # _property_mapping = 'attrs' # __map = property(lambda self: getattr(self, self._property_mapping)) # def __getitem__(self, keyobj): # return self.__map.get(keyobj.name, None) # # def __setitem__(self, keyobj, value): # attr_obj = self.__map.get(keyobj.name, None) # if attr_obj is None: # self.__map[keyobj.name] = self.attr_class(key=keyobj, value=value) # else: # pass import unittest from sqlalchemy import create_engine from sqlalchemy.exc import IntegrityError class TestSchema(unittest.TestCase): def setUp(self): #engine = create_engine(oracle_db_path, echo=False) engine = create_engine('sqlite://', echo=False) metadata.drop_all(engine) metadata.create_all(engine, checkfirst=False) Session.configure(bind=engine) def tearDown(self): Session.remove() def test_enum_values(self): session = Session() # create an attribute and some values a = MyVertical.key_class('gender', 'enum', 'patient gender') session.add(a) a.set_enum('female', 'a woman') a.set_enum('man', 'a man') a.set_enum('transgender', 'a transgender dude') session.commit() # check the add worked self.assertTrue('man' in a.enum_values) self.assertTrue('female' in a.enum_values) self.assertTrue('transgender' in a.enum_values) # try changing a description a.enum_values['man'].description = 'george' session.commit() # try changing a value a.enum_values['man'].value = 'bubba' session.commit() self.assertTrue(a.enum_values['bubba'].description == 'george') # try deleting something del a.enum_values['transgender'] session.commit() # verify delete worked self.assertTrue('bubba' in a.enum_values) self.assertTrue('female' in a.enum_values) self.assertFalse('transgender' in a.enum_values) return def test_add_keys(self): session = Session() # create an attribute and some values a = MyVertical.key_class('age', 'int', 'patient age') session.add(a) session.commit() # check the add worked self.assertTrue(a == session.query(MyVertical.key_class).filter_by(name='age').one()) self.assertTrue(a.type == 'int') # now bool a = MyVertical.key_class('fat', 'bool', 'is patient fat') session.add(a) session.commit() # check the add worked self.assertTrue(a == session.query(MyVertical.key_class).filter_by(name='fat').one()) self.assertTrue(a.type == 'bool') def test_sample_attrs(self): session = Session() # create generic attributes myint = MyVertical.key_class('age', 'int', 'patient age') mybool = MyVertical.key_class('smoking', 'bool', 'does the guy smoke') myfloat = MyVertical.key_class('ratio', 'float', 'fraction') session.add_all([myint, mybool, myfloat]) # create some enum attributes gender = MyVertical.key_class('gender', 'enum', 'patient gender') gender.enum_values['male'] = MyVertical.enum_value_class('male', 'a man') gender.enum_values['female'] = MyVertical.enum_value_class('female', 'a woman') session.add(gender) cancer = MyVertical.key_class('cancer_status', 'enum', 'whether or not patient has cancer') cancer.enum_values['no'] = MyVertical.enum_value_class('no', 'no cancer present') cancer.enum_values['yes'] = MyVertical.enum_value_class('yes', 'cancer present') session.add(cancer) blood = MyVertical.key_class('blood_type', 'enum', 'patient blood type') blood.enum_values['A'] = MyVertical.enum_value_class('A', 'A type') blood.enum_values['B'] = MyVertical.enum_value_class('B', 'B type') blood.enum_values['C'] = MyVertical.enum_value_class('C', 'C type') blood.enum_values['D'] = MyVertical.enum_value_class('D', 'D type') session.add(blood) # create objects obj1 = MyVertical('obj1') obj2 = MyVertical('obj2') obj3 = MyVertical('obj3') session.add_all([obj1, obj2, obj3]) session.commit() # assign attributes to objects obj1.attrs[myint] = 10 obj1.attrs[mybool] = True obj1.attrs[myfloat] = 3.14 session.commit() self.assertTrue(obj1.attrs[myint] == 10) self.assertTrue(obj1.attrs[mybool] == True) self.assertTrue(obj1.attrs[myfloat] == 3.14) # set enum attributes obj1.attrs[blood] = blood['A'] obj1.attrs[cancer] = cancer['yes'] obj1.attrs[gender] = gender['female'] obj2.attrs[blood] = blood['B'] obj2.attrs[cancer] = cancer['no'] obj2.attrs[gender] = gender['female'] session.commit() # verify self.assertTrue(obj1.attrs[blood].value == 'A') self.assertTrue(obj1.attrs[cancer].value == 'yes') self.assertTrue(obj1.attrs[gender].value == 'female') # change a setting obj1.attrs[blood] = blood['B'] session.commit() self.assertTrue(obj1.attrs[blood].value == 'B') # change a key name and value gender.name = 'sex' gender['female'].value = 'woman' session.commit() # verify it did not change things self.assertTrue('female' not in gender) self.assertTrue('woman' in gender) self.assertTrue(obj1.attrs[gender].value == 'woman') self.assertTrue(obj2.attrs[gender].value == 'woman') # try to change to an existing enum value gender['woman'].value = 'male' self.assertRaises(IntegrityError, session.commit) session.rollback() # delete attribute and verify del obj1.attrs[blood] session.commit() blood = session.query(MyVertical.key_class).filter_by(name='blood_type').one() self.assertFalse(blood in obj1.attrs) self.assertTrue(cancer in obj1.attrs) self.assertTrue(gender in obj1.attrs) self.assertTrue(blood in obj2.attrs) def test_enum_value_updates(self): session = Session() # create enum attributes with same values cancer = MyVertical.key_class('cancer_status', 'enum', 'whether or not patient has cancer') cancer.enum_values['no'] = MyVertical.enum_value_class('no', 'no cancer present') cancer.enum_values['yes'] = MyVertical.enum_value_class('yes', 'cancer present') session.add(cancer) smoking = MyVertical.key_class('smoking', 'enum', 'whether or not patient smokes') smoking.enum_values['no'] = MyVertical.enum_value_class('no', 'no smoke') smoking.enum_values['yes'] = MyVertical.enum_value_class('yes', 'yes smoke') session.add(smoking) # create objects obj1 = MyVertical('obj1') # set enum attributes obj1.attrs[cancer] = cancer['yes'] obj1.attrs[smoking] = smoking['yes'] session.commit() self.assertTrue(obj1.attrs[cancer] == cancer['yes']) self.assertTrue(obj1.attrs[smoking] == smoking['yes']) # change self.assertRaises(KeyError, smoking.__getitem__, 'maybe') obj1.attrs[smoking] = smoking['no'] session.commit() self.assertTrue(obj1.attrs[cancer] == cancer['yes']) self.assertTrue(obj1.attrs[smoking] == smoking['no']) return class TestInterface(unittest.TestCase): def setUp(self): #engine = create_engine(oracle_db_path, echo=False) engine = create_engine('sqlite://', echo=False) metadata.drop_all(engine) metadata.create_all(engine, checkfirst=False) Session.configure(bind=engine) def tearDown(self): Session.remove() def test_enums(self): session = Session() # create object bob = MyVertical('bob') session.add(bob) age_key = bob.set_key('age', 'enum', description='bob age', values=None) session.add(age_key) session.commit() # try setting some values age_vals = ['baby', 'child', 'teen', 'adult'] age_key.set_enums(age_vals) session.commit() self.assertTrue(set(age_vals) == set(MyVertical.query_key('age').get_enums())) session.commit() # add object with values size_vals = ['xs', 's', 'med', 'large'] size_key = MyVertical.set_key('size', 'enum', description='the size', values=size_vals) session.add(size_key) session.commit() self.assertTrue(set(size_vals) == set(MyVertical.query_key('size').get_enums())) self.assertFalse(set(age_vals) == set(MyVertical.query_key('size').get_enums())) # delete values del age_key['child'] age_vals.remove('child') session.commit() self.assertTrue(set(age_vals) == set(MyVertical.query_key('age').get_enums())) def test_attr_keys(self): session = Session() # add attributes a = MyVertical.set_key('a', key_type='enum', description='attr a') b = MyVertical.set_key('b', key_type='enum', description='attr b') c = MyVertical.set_key('c', key_type='enum', description='attr c') session.add_all([a, b, c]) session.commit() # verify attrnames = set([x.name for x in MyVertical.query_keys()]) self.assertTrue(set(['a', 'b', 'c']) == attrnames) # try adding duplicate self.assertRaises(SequelDBError, MyVertical.create_key, 'a', 'enum', 'duplicate a') # try updating a key MyVertical.update_key('b', description="update attr b") session.commit() b = MyVertical.query_key('b') self.assertTrue(b.description == 'update attr b') # try changing an attribute b.name = 'b2' b.description = 'attr b2' session.commit() self.assertTrue(MyVertical.query_key('b') == None) b2 = MyVertical.query_key('b2') self.assertTrue(b2.name == 'b2') self.assertTrue(b2.description == 'attr b2') # try deleting and re-adding attributes MyVertical.del_keys(['a', 'c']) session.commit() attrs = list(MyVertical.query_keys()) self.assertTrue(len(attrs) == 1) self.assertTrue(attrs[0].name == 'b2') def test_attr_values(self): session = Session() a = MyVertical.set_key('a', 'enum', description='attr a') session.add(a) # add attribute values a_vals = [('w', 'what'), 'x', ('y', 'yo'), 'z'] a.set_enums(a_vals) session.commit() self.assertTrue(set(a.get_enums()) == set(['w', 'x', 'y', 'z'])) self.assertTrue(a['w'].description == 'what') self.assertTrue(a['x'].description == None) self.assertTrue(a['y'].description == 'yo') self.assertTrue(a['z'].description == None) # modify attribute values a['x'].value = 'CHANGED' self.assertTrue('x' in a) # need to commit to re-instrument session.commit() self.assertTrue('x' not in a) # delete and re-add del a['CHANGED'] a.set_enum('x', description='x is back') session.commit() self.assertTrue('x' in a) self.assertTrue(a['x'].description == 'x is back') return def test_vertical_objects(self): session = Session() # add objects obj1 = MyVertical('vcap') obj2 = MyVertical('lncap') obj3 = MyVertical('am28') session.add_all([obj1, obj2, obj3]) session.commit() session.expunge_all() names = set([obj.name for obj in session.query(MyVertical)]) self.assertTrue(set(['vcap', 'lncap', 'am28']) == names) return def test_vertical_attrs(self): session = Session() # add objects vcap = MyVertical('vcap') lncap = MyVertical('lncap') am28 = MyVertical('am28') session.add_all([vcap, lncap, am28]) session.commit() # add attributes a = MyVertical.set_key('a', 'enum', description='enum attr a', values=[('a1', 'one'), ('a2', 'two'), 'a3', 'a4', 'a5']) b = MyVertical.set_key('b', 'int', description='int attr b') c = MyVertical.set_key('c', 'float', description='float attr c') session.add_all([a, b, c]) session.commit() # associate attributes with objects vcap_kvpairs = [(a, a['a1']), (b, 2), (c, 5.1)] lncap_kvpairs = [(a, a['a2']), (b, 1), (c, 0.5)] am28_kvpairs = [(a, a['a3']), (b, 4), (c, 2.0)] vcap.attrs.update(vcap_kvpairs) lncap.attrs.update(lncap_kvpairs) am28.attrs.update(am28_kvpairs) session.commit() # check self.assertTrue(set(vcap_kvpairs) == set(vcap.attrs.items())) self.assertTrue(set(lncap_kvpairs) == set(lncap.attrs.items())) self.assertTrue(set(am28_kvpairs) == set(am28.attrs.items())) # modify sample attributes a = MyVertical.query_key('a') b = MyVertical.query_key('b') c = MyVertical.query_key('c') # change attribute value vcap.attrs[a] = a['a5'] lncap.attrs[a] = a['a5'] am28.attrs[a] = a['a5'] session.commit() vcap_kvpairs[0] = (a, a['a5']) lncap_kvpairs[0] = (a, a['a5']) am28_kvpairs[0] = (a, a['a5']) self.assertTrue(set(vcap_kvpairs) == set(vcap.attrs.items())) self.assertTrue(set(lncap_kvpairs) == set(lncap.attrs.items())) self.assertTrue(set(am28_kvpairs) == set(am28.attrs.items())) # rename an attribute and check if name propagates to all samples a.name = 'renameA' session.commit() a_attr_names = [k.name for k in vcap.attrs.keys()] self.assertTrue('a' not in a_attr_names) self.assertTrue('renameA' in a_attr_names) vcap_kvpairs[0] = (a, a['a5']) lncap_kvpairs[0] = (a, a['a5']) am28_kvpairs[0] = (a, a['a5']) self.assertTrue(set(vcap_kvpairs) == set(vcap.attrs.items())) # rename an attribute value and check if value propagates to all samples a['a5'].value = 'renameA5' session.commit() self.assertTrue('a5' not in a) self.assertTrue('renameA5' in a) self.assertTrue(vcap.attrs[a].value == 'renameA5') self.assertTrue(lncap.attrs[a].value == 'renameA5') self.assertTrue(am28.attrs[a].value == 'renameA5') # delete an attribute value and check if deletion propagates to sample attributes del a['renameA5'] vcap_kvpairs.pop(0) lncap_kvpairs.pop(0) am28_kvpairs.pop(0) session.commit() self.assertTrue(set(vcap_kvpairs) == set(vcap.attrs.items())) self.assertTrue(set(lncap_kvpairs) == set(lncap.attrs.items())) self.assertTrue(set(am28_kvpairs) == set(am28.attrs.items())) # delete an attribute key and check if deletion propagates to sample attributes MyVertical.del_keys(['b']) session.commit() vcap_kvpairs.pop(0) lncap_kvpairs.pop(0) am28_kvpairs.pop(0) self.assertTrue(set(vcap_kvpairs) == set(vcap.attrs.items())) self.assertTrue(set(lncap_kvpairs) == set(lncap.attrs.items())) self.assertTrue(set(am28_kvpairs) == set(am28.attrs.items())) if __name__ == '__main__': class MyVertical(Base, VerticalAttrMixin): ''' must inherit from declarative base (Base) in order for class to work must define __metaclass__ = VerticalAttrMetaclass and set __tablename__ to some user-defined table name ''' __metaclass__ = VerticalAttrMetaclass __tablename__ = 'myverticals' id = Column(Integer, Sequence('myvertical_id_seq'), primary_key=True) name = Column(String(50)) def __init__(self, name): super(MyVertical, self).__init__() self.name = name def __repr__(self): return "<MyVertical(name='%s')>" % self.name unittest.main() -- You received this message because you are subscribed to the Google Groups "sqlalchemy" group. To post to this group, send email to sqlalch...@googlegroups.com. To unsubscribe from this group, send email to sqlalchemy+unsubscr...@googlegroups.com. For more options, visit this group at http://groups.google.com/group/sqlalchemy?hl=en.