polymorph2.py

SQLAlchemy. 经典的Python ORM框架。学习必看。

            else:
                st=dead
            session.save(Manager(name="M%d" % i,category="YYYYYYYYY",status=st))
            session.save(Engineer(name="E%d" % i,field="X",status=st))

        session.flush()

        # get E4
        engineer4 = session.query(engineer_mapper).filter_by(name="E4").one()

        # create 2 cars for E4, one active and one dead
        car1 = Car(employee=engineer4,status=active)
        car2 = Car(employee=engineer4,status=dead)
        session.save(car1)
        session.save(car2)
        session.flush()

        # test these twice because theres caching involved, as well previous issues that modified the polymorphic union
        for x in range(0, 2):
            r = session.query(Person).filter(people.c.name.like('%2')).join('status').filter_by(name="active")
            assert str(list(r)) == "[Manager M2, category YYYYYYYYY, status Status active, Engineer E2, field X, status Status active]"
            r = session.query(Engineer).join('status').filter(people.c.name.in_(['E2', 'E3', 'E4', 'M4', 'M2', 'M1']) & (status.c.name=="active"))
            assert str(list(r)) == "[Engineer E2, field X, status Status active, Engineer E3, field X, status Status active]"
            # this test embeds the original polymorphic union (employee_join) fully
            # into the WHERE criterion, using a correlated select. ticket #577 tracks
            # that Query's adaptation of the WHERE clause does not dig into the
            # mapped selectable itself, which permanently breaks the mapped selectable.
            r = session.query(Person).filter(exists([Car.c.owner], Car.c.owner==employee_join.c.person_id))
            assert str(list(r)) == "[Engineer E4, field X, status Status dead]"

class MultiLevelTest(ORMTest):
    def define_tables(self, metadata):
        global table_Employee, table_Engineer, table_Manager
        table_Employee = Table( 'Employee', metadata,
            Column( 'name', type_= String(100), ),
            Column( 'id', primary_key= True, type_= Integer, ),
            Column( 'atype', type_= String(100), ),
        )

        table_Engineer = Table( 'Engineer', metadata,
            Column( 'machine', type_= String(100), ),
            Column( 'id', Integer, ForeignKey( 'Employee.id', ), primary_key= True, ),
        )

        table_Manager = Table( 'Manager', metadata,
            Column( 'duties', type_= String(100), ),
            Column( 'id', Integer, ForeignKey( 'Engineer.id', ), primary_key= True, ),
        )
    def test_threelevels(self):
        class Employee( object):
            def set( me, **kargs):
                for k,v in kargs.iteritems(): setattr( me, k, v)
                return me
            def __str__(me): return str(me.__class__.__name__)+':'+str(me.name)
            __repr__ = __str__
        class Engineer( Employee): pass
        class Manager( Engineer): pass

        pu_Employee = polymorphic_union( {
                    'Manager':  table_Employee.join( table_Engineer).join( table_Manager),
                    'Engineer': select([table_Employee, table_Engineer.c.machine], table_Employee.c.atype == 'Engineer', from_obj=[table_Employee.join(table_Engineer)]),
                    'Employee': table_Employee.select( table_Employee.c.atype == 'Employee'),
                }, None, 'pu_employee', )

#        pu_Employee = polymorphic_union( {
#                    'Manager':  table_Employee.join( table_Engineer).join( table_Manager),
#                    'Engineer': table_Employee.join(table_Engineer).select(table_Employee.c.atype == 'Engineer'),
#                    'Employee': table_Employee.select( table_Employee.c.atype == 'Employee'),
#                }, None, 'pu_employee', )

        mapper_Employee = mapper( Employee, table_Employee,
                    polymorphic_identity= 'Employee',
                    polymorphic_on= pu_Employee.c.atype,
                    select_table= pu_Employee,
                )

        pu_Engineer = polymorphic_union( {
                    'Manager':  table_Employee.join( table_Engineer).join( table_Manager),
                    'Engineer': select([table_Employee, table_Engineer.c.machine], table_Employee.c.atype == 'Engineer', from_obj=[table_Employee.join(table_Engineer)]),
                }, None, 'pu_engineer', )
        mapper_Engineer = mapper( Engineer, table_Engineer,
                    inherit_condition= table_Engineer.c.id == table_Employee.c.id,
                    inherits= mapper_Employee,
                    polymorphic_identity= 'Engineer',
                    polymorphic_on= pu_Engineer.c.atype,
                    select_table= pu_Engineer,
                )

        mapper_Manager = mapper( Manager, table_Manager,
                    inherit_condition= table_Manager.c.id == table_Engineer.c.id,
                    inherits= mapper_Engineer,
                    polymorphic_identity= 'Manager',
                )

        a = Employee().set( name= 'one')
        b = Engineer().set( egn= 'two', machine= 'any')
        c = Manager().set( name= 'head', machine= 'fast', duties= 'many')

        session = create_session()
        session.save(a)
        session.save(b)
        session.save(c)
        session.flush()
        assert set(session.query(Employee).all()) == set([a,b,c])
        assert set(session.query( Engineer).all()) == set([b,c])
        assert session.query( Manager).all() == [c]

class ManyToManyPolyTest(ORMTest):
    def define_tables(self, metadata):
        global base_item_table, item_table, base_item_collection_table, collection_table
        base_item_table = Table(
            'base_item', metadata,
            Column('id', Integer, primary_key=True),
            Column('child_name', String(255), default=None))

        item_table = Table(
            'item', metadata,
            Column('id', Integer, ForeignKey('base_item.id'), primary_key=True),
            Column('dummy', Integer, default=0)) # Dummy column to avoid weird insert problems

        base_item_collection_table = Table(
            'base_item_collection', metadata,
            Column('item_id', Integer, ForeignKey('base_item.id')),
            Column('collection_id', Integer, ForeignKey('collection.id')))

        collection_table = Table(
            'collection', metadata,
            Column('id', Integer, primary_key=True),
            Column('name', Unicode(255)))

    def test_pjoin_compile(self):
        """test that remote_side columns in the secondary join table arent attempted to be
        matched to the target polymorphic selectable"""
        class BaseItem(object): pass
        class Item(BaseItem): pass
        class Collection(object): pass
        item_join = polymorphic_union( {
            'BaseItem':base_item_table.select(base_item_table.c.child_name=='BaseItem'),
            'Item':base_item_table.join(item_table),
            }, None, 'item_join')

        mapper(
            BaseItem, base_item_table,
            select_table=item_join,
            polymorphic_on=base_item_table.c.child_name,
            polymorphic_identity='BaseItem',
            properties=dict(collections=relation(Collection, secondary=base_item_collection_table, backref="items")))

        mapper(
            Item, item_table,
            inherits=BaseItem,
            polymorphic_identity='Item')

        mapper(Collection, collection_table)

        class_mapper(BaseItem)

class CustomPKTest(ORMTest):
    def define_tables(self, metadata):
        global t1, t2
        t1 = Table('t1', metadata,
            Column('id', Integer, primary_key=True),
            Column('type', String(30), nullable=False),
            Column('data', String(30)))
        # note that the primary key column in t2 is named differently
        t2 = Table('t2', metadata,
            Column('t2id', Integer, ForeignKey('t1.id'), primary_key=True),
            Column('t2data', String(30)))

    def test_custompk(self):
        """test that the primary_key attribute is propigated to the polymorphic mapper"""

        class T1(object):pass
        class T2(T1):pass

        # create a polymorphic union with the select against the base table first.
        # with the join being second, the alias of the union will
        # pick up two "primary key" columns.  technically the alias should have a
        # 2-col pk in any case but the leading select has a NULL for the "t2id" column
        d = util.OrderedDict()
        d['t1'] = t1.select(t1.c.type=='t1')
        d['t2'] = t1.join(t2)
        pjoin = polymorphic_union(d, None, 'pjoin')

        mapper(T1, t1, polymorphic_on=t1.c.type, polymorphic_identity='t1', select_table=pjoin, primary_key=[pjoin.c.id])
        mapper(T2, t2, inherits=T1, polymorphic_identity='t2')
        print [str(c) for c in class_mapper(T1).primary_key]
        ot1 = T1()
        ot2 = T2()
        sess = create_session()
        sess.save(ot1)
        sess.save(ot2)
        sess.flush()
        sess.clear()

        # query using get(), using only one value.  this requires the select_table mapper
        # has the same single-col primary key.
        assert sess.query(T1).get(ot1.id).id == ot1.id

        ot1 = sess.query(T1).get(ot1.id)
        ot1.data = 'hi'
        sess.flush()

    def test_pk_collapses(self):
        """test that a composite primary key attribute formed by a join is "collapsed" into its
        minimal columns"""

        class T1(object):pass
        class T2(T1):pass

        # create a polymorphic union with the select against the base table first.
        # with the join being second, the alias of the union will
        # pick up two "primary key" columns.  technically the alias should have a
        # 2-col pk in any case but the leading select has a NULL for the "t2id" column
        d = util.OrderedDict()
        d['t1'] = t1.select(t1.c.type=='t1')
        d['t2'] = t1.join(t2)
        pjoin = polymorphic_union(d, None, 'pjoin')

        mapper(T1, t1, polymorphic_on=t1.c.type, polymorphic_identity='t1', select_table=pjoin)
        mapper(T2, t2, inherits=T1, polymorphic_identity='t2')
        assert len(class_mapper(T1).primary_key) == 1
        assert len(class_mapper(T1).get_select_mapper().compile().primary_key) == 1

        print [str(c) for c in class_mapper(T1).primary_key]
        ot1 = T1()
        ot2 = T2()
        sess = create_session()
        sess.save(ot1)
        sess.save(ot2)
        sess.flush()
        sess.clear()

        # query using get(), using only one value.  this requires the select_table mapper
        # has the same single-col primary key.
        assert sess.query(T1).get(ot1.id).id == ot1.id

        ot1 = sess.query(T1).get(ot1.id)
        ot1.data = 'hi'
        sess.flush()

class InheritingEagerTest(ORMTest):
    def define_tables(self, metadata):
        global people, employees, tags, peopleTags

        people = Table('people', metadata,
                           Column('id', Integer, primary_key=True),
                           Column('_type', String(30), nullable=False),
                          )


        employees = Table('employees', metadata,
                         Column('id', Integer, ForeignKey('people.id'),primary_key=True),
                        )

        tags = Table('tags', metadata,
                           Column('id', Integer, primary_key=True),
                           Column('label', String(50), nullable=False),
                       )

        peopleTags = Table('peopleTags', metadata,
                               Column('person_id', Integer,ForeignKey('people.id')),
                               Column('tag_id', Integer,ForeignKey('tags.id')),
                         )

    def test_basic(self):
        """test that Query uses the full set of mapper._eager_loaders when generating SQL"""

        class Person(fixtures.Base):
            pass

        class Employee(Person):
           def __init__(self, name='bob'):
               self.name = name

        class Tag(fixtures.Base):
           def __init__(self, label):
               self.label = label

        mapper(Person, people, polymorphic_on=people.c._type,polymorphic_identity='person', properties={
            'tags': relation(Tag, secondary=peopleTags,backref='people', lazy=False)
        })
        mapper(Employee, employees, inherits=Person,polymorphic_identity='employee')
        mapper(Tag, tags)

        session = create_session()

        bob = Employee()
        session.save(bob)

        tag = Tag('crazy')
        bob.tags.append(tag)

        tag = Tag('funny')
        bob.tags.append(tag)
        session.flush()

        session.clear()
        # query from Employee with limit, query needs to apply eager limiting subquery
        instance = session.query(Employee).filter_by(id=1).limit(1).first()
        assert len(instance.tags) == 2

class MissingPolymorphicOnTest(ORMTest):
    def define_tables(self, metadata):
        global tablea, tableb, tablec, tabled
        tablea = Table('tablea', metadata, 
            Column('id', Integer, primary_key=True),
            Column('adata', String(50)),
            )
        tableb = Table('tableb', metadata, 
            Column('id', Integer, primary_key=True),
            Column('aid', Integer, ForeignKey('tablea.id')),
            Column('data', String(50)),
            )
        tablec = Table('tablec', metadata, 
            Column('id', Integer, ForeignKey('tablea.id'), primary_key=True),
            Column('cdata', String(50)),
            )
        tabled = Table('tabled', metadata, 
            Column('id', Integer, ForeignKey('tablec.id'), primary_key=True),
            Column('ddata', String(50)),
            )
            
    def test_polyon_col_setsup(self):
        class A(fixtures.Base):
            pass
        class B(fixtures.Base):
            pass
        class C(A):
            pass
        class D(C):
            pass
            
        poly_select = select([tablea, tableb.c.data.label('discriminator')], from_obj=tablea.join(tableb)).alias('poly')
        
        mapper(B, tableb)
        mapper(A, tablea, select_table=poly_select, polymorphic_on=poly_select.c.discriminator, properties={
            'b':relation(B, uselist=False)
        })
        mapper(C, tablec, inherits=A,polymorphic_identity='c')
        mapper(D, tabled, inherits=C, polymorphic_identity='d')
        
        c = C(cdata='c1', adata='a1', b=B(data='c'))
        d = D(cdata='c2', adata='a2', ddata='d2', b=B(data='d'))
        sess = create_session()
        sess.save(c)
        sess.save(d)
        sess.flush()
        sess.clear()
        self.assertEquals(sess.query(A).all(), [C(cdata='c1', adata='a1'), D(cdata='c2', adata='a2', ddata='d2')])
        
if __name__ == "__main__":
    testenv.main()