Extending and automating the vertical dictionary paradigm
29 views
Skip to first unread message
Choy
Aug 12, 2010, 5:08:03 PM8/12/10
to sqlalchemy
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()
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()
Reply all
Reply to author
Forward
0 new messages