Commits

Mike Bayer committed 624495d

integrating Jonathan LaCour's declarative layer

Comments (0)

Files changed (2)

lib/sqlalchemy/ext/activemapper.py

+from sqlalchemy             import objectstore, create_engine, assign_mapper, relation, mapper
+from sqlalchemy             import and_, or_
+from sqlalchemy             import Table, Column
+from sqlalchemy.ext.proxy   import ProxyEngine
+
+import inspect
+
+
+#
+# the "proxy" to the database engine... this can be swapped out at runtime
+#
+engine = ProxyEngine()
+
+
+
+#
+# declarative column declaration - this is so that we can infer the colname
+#
+class column(object):
+    def __init__(self, coltype, colname=None, foreign_key=None, primary_key=False):
+        self.coltype     = coltype
+        self.colname     = colname
+        self.foreign_key = foreign_key
+        self.primary_key = primary_key
+
+
+
+#
+# declarative relationship declaration
+#
+class relationship(object):
+    def __init__(self, classname, colname=None, backref=None, private=False, lazy=True, uselist=True):
+        self.classname = classname
+        self.colname   = colname
+        self.backref   = backref
+        self.private   = private
+        self.lazy      = lazy
+        self.uselist   = uselist
+
+
+class one_to_many(relationship):
+    def __init__(self, classname, colname=None, backref=None, private=False, lazy=True):
+        relationship.__init__(self, classname, colname, backref, private, lazy, uselist=True)
+
+
+class one_to_one(relationship):
+    def __init__(self, classname, colname=None, backref=None, private=False, lazy=True):
+        relationship.__init__(self, classname, colname, backref, private, lazy, uselist=False)
+
+
+
+# 
+# SQLAlchemy metaclass and superclass that can be used to do SQLAlchemy 
+# mapping in a declarative way, along with a function to process the 
+# relationships between dependent objects as they come in, without blowing
+# up if the classes aren't specified in a proper order
+# 
+
+__deferred_classes__  = []
+def process_relationships(klass, was_deferred=False):
+    defer = False
+    for propname, reldesc in klass.relations.items():
+        if not reldesc.classname in ActiveMapperMeta.classes:
+            if not was_deferred: __deferred_classes__.append(klass)
+            defer = True
+    
+    if not defer:
+        relations = {}
+        for propname, reldesc in klass.relations.items():
+            relclass = ActiveMapperMeta.classes[reldesc.classname]
+            relations[propname] = relation(relclass, 
+                                           backref=reldesc.backref, 
+                                           private=reldesc.private, 
+                                           lazy=reldesc.lazy, 
+                                           uselist=reldesc.uselist)
+        assign_mapper(klass, klass.table, properties=relations)
+        if was_deferred: __deferred_classes__.remove(klass)
+    
+    if not was_deferred:
+        for deferred_class in __deferred_classes__:
+            process_relationships(deferred_class, was_deferred=True)
+
+
+
+class ActiveMapperMeta(type):
+    classes = {}
+    
+    def __init__(cls, clsname, bases, dict):
+        table_name = clsname.lower()
+        columns    = []
+        relations  = {}
+        
+        if 'mapping' in dict:
+            members = inspect.getmembers(dict.get('mapping'))
+            for name, value in members:
+                if name == '__table__':
+                    table_name = value
+                    continue
+                
+                if name.startswith('__'): continue
+                
+                if isinstance(value, column):
+                    if value.foreign_key:
+                        col = Column(value.colname or name, 
+                                     value.coltype,
+                                     value.foreign_key, 
+                                     primary_key=value.primary_key)
+                    else:
+                        col = Column(value.colname or name,
+                                     value.coltype,
+                                     primary_key=value.primary_key)
+                    columns.append(col)
+                    continue
+                
+                if isinstance(value, relationship):
+                    relations[name] = value
+            
+            cls.table = Table(table_name, engine, *columns)
+            assign_mapper(cls, cls.table)
+            cls.relations = relations
+            ActiveMapperMeta.classes[clsname] = cls
+            
+            process_relationships(cls)
+        
+        super(ActiveMapperMeta, cls).__init__(clsname, bases, dict)
+
+
+class ActiveMapper(object):
+    __metaclass__ = ActiveMapperMeta
+    
+    def set(self, **kwargs):
+        for key, value in kwargs.items():
+            setattr(self, key, value)
+
+
+#
+# a utility function to create all tables for all ActiveMapper classes
+#
+
+def create_tables():
+    for klass in ActiveMapperMeta.classes.values():
+        klass.table.create()

test/activemapper.py

+from activemapper           import ActiveMapper, column, one_to_many, one_to_one
+from sqlalchemy             import objectstore
+from sqlalchemy             import and_, or_
+from sqlalchemy             import ForeignKey, String, Integer, DateTime
+from datetime               import datetime
+
+import unittest
+import activemapper
+
+#
+# application-level model objects
+#
+
+class Person(ActiveMapper):
+    class mapping:
+        id          = column(Integer, primary_key=True)
+        full_name   = column(String)
+        first_name  = column(String)
+        middle_name = column(String)
+        last_name   = column(String)
+        birth_date  = column(DateTime)
+        ssn         = column(String)
+        gender      = column(String)
+        home_phone  = column(String)
+        cell_phone  = column(String)
+        work_phone  = column(String)
+        prefs_id    = column(Integer, foreign_key=ForeignKey('preferences.id'))
+        addresses   = one_to_many('Address', colname='person_id', backref='person')
+        preferences = one_to_one('Preferences', colname='pref_id', backref='person')
+    
+    def __str__(self):
+        s =  '%s\n' % self.full_name
+        s += '  * birthdate: %s\n' % (self.birth_date or 'not provided')
+        s += '  * fave color: %s\n' % (self.preferences.favorite_color or 'Unknown')
+        s += '  * personality: %s\n' % (self.preferences.personality_type or 'Unknown')
+        
+        for address in self.addresses:
+            s += '  * address: %s\n' % address.address_1
+            s += '             %s, %s %s\n' % (address.city, address.state, address.postal_code)
+        
+        return s
+
+
+class Preferences(ActiveMapper):
+    class mapping:
+        __table__        = 'preferences'
+        id               = column(Integer, primary_key=True)
+        favorite_color   = column(String)
+        personality_type = column(String)
+
+
+class Address(ActiveMapper):
+    class mapping:
+        id          = column(Integer, primary_key=True)
+        type        = column(String)
+        address_1   = column(String)
+        city        = column(String)
+        state       = column(String)
+        postal_code = column(String)
+        person_id   = column(Integer, foreign_key=ForeignKey('person.id'))
+
+
+
+class testcase(unittest.TestCase):    
+    
+    def tearDown(self):
+        people = Person.select()
+        for person in people: person.delete()
+        
+        addresses = Address.select()
+        for address in addresses: address.delete()
+        
+        preferences = Preferences.select()
+        for preference in preferences: preference.delete()
+        
+        objectstore.commit()
+        objectstore.clear()
+    
+    def create_person_one(self):
+        # create a person
+        p1 = Person(
+                full_name='Jonathan LaCour',
+                birth_date=datetime(1979, 10, 12),
+                preferences=Preferences(
+                                favorite_color='Green',
+                                personality_type='ENTP'
+                            ),
+                addresses=[
+                    Address(
+                        address_1='123 Some Great Road.',
+                        city='Atlanta',
+                        state='GA',
+                        postal_code='30338'
+                    ),
+                    Address(
+                        address_1='435 Franklin Road.',
+                        city='Atlanta',
+                        state='GA',
+                        postal_code='30342'
+                    )
+                ]
+             )
+        return p1
+    
+    
+    def create_person_two(self):
+        p2 = Person(
+                full_name='Lacey LaCour',
+                addresses=[
+                    Address(
+                        address_1='123 Some Great Road.',
+                        city='Atlanta',
+                        state='GA',
+                        postal_code='30338'
+                    ),
+                    Address(
+                        address_1='200 Main Street',
+                        city='Roswell',
+                        state='GA',
+                        postal_code='30075'
+                    )
+                ]
+             )
+        # I don't like that I have to do this... and putting
+        # a "self.preferences = Preferences()" into the __init__
+        # of Person also doens't seem to fix this
+        p2.preferences = Preferences()
+        
+        return p2
+    
+    
+    def test_create(self):
+        p1 = self.create_person_one()
+        
+        objectstore.commit()
+        objectstore.clear()
+        
+        results = Person.select()
+        
+        self.assertEquals(len(results), 1)
+        
+        person = results[0]
+        self.assertEquals(person.id, p1.id)
+        self.assertEquals(len(person.addresses), 2)
+        self.assertEquals(person.addresses[0].postal_code, '30338')
+    
+    
+    def test_delete(self):
+        p1 = self.create_person_one()
+        
+        objectstore.commit()
+        objectstore.clear()
+        
+        results = Person.select()
+        self.assertEquals(len(results), 1)
+        
+        results[0].delete()
+        objectstore.commit()
+        objectstore.clear()
+        
+        results = Person.select()
+        self.assertEquals(len(results), 0)
+    
+    
+    def test_multiple(self):
+        p1 = self.create_person_one()
+        p2 = self.create_person_two()
+        
+        objectstore.commit()
+        objectstore.clear()
+        
+        # select and make sure we get back two results
+        people = Person.select()
+        self.assertEquals(len(people), 2)
+                
+        # make sure that our backwards relationships work
+        self.assertEquals(people[0].addresses[0].person.id, p1.id)
+        self.assertEquals(people[1].addresses[0].person.id, p2.id)
+        
+        # try a more complex select
+        results = Person.select(
+            or_(
+                and_(
+                    Address.c.person_id == Person.c.id,
+                    Address.c.postal_code.like('30075')
+                ),
+                and_(
+                    Person.c.prefs_id == Preferences.c.id,
+                    Preferences.c.favorite_color == 'Green'
+                )
+            )
+        )
+        self.assertEquals(len(results), 2)
+        
+    
+    def test_oneway_backref(self):
+        # FIXME: I don't know why, but it seems that my backwards relationship
+        #        on preferences still ends up being a list even though I pass
+        #        in uselist=False...
+        p1 = self.create_person_one()
+        self.assertEquals(p1.preferences.person, p1)
+        p1.delete()
+        
+        objectstore.commit()
+        objectstore.clear()
+    
+    
+    def test_select_by(self):
+        # FIXME: either I don't understand select_by, or it doesn't work.
+        
+        p1 = self.create_person_one()
+        p2 = self.create_person_two()
+        
+        objectstore.commit()
+        objectstore.clear()
+        
+        results = Person.select_by(
+            Address.c.postal_code.like('30075')
+        )
+        self.assertEquals(len(results), 1)
+
+
+    
+if __name__ == '__main__':
+    # go ahead and setup the database connection, and create the tables
+    activemapper.engine.connect('sqlite:///', echo=False)
+    activemapper.create_tables()
+    
+    # launch the unit tests
+    unittest.main()