selectable.py

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

        a = Table('a', meta, Column('id', Integer, primary_key=True))
        b = Table('b', meta, Column('id', Integer, ForeignKey('a.id'), primary_key=True))
        c = Table('c', meta, Column('id', Integer, ForeignKey('b.id'), primary_key=True))
        d = Table('d', meta, Column('id', Integer, ForeignKey('c.id'), primary_key=True))

        assert c.c.id.references(b.c.id)
        assert not d.c.id.references(a.c.id)

        assert list(a.join(b).primary_key) == [a.c.id]
        assert list(b.join(c).primary_key) == [b.c.id]
        assert list(a.join(b).join(c).primary_key) == [a.c.id]
        assert list(b.join(c).join(d).primary_key) == [b.c.id]
        assert list(d.join(c).join(b).primary_key) == [b.c.id]
        assert list(a.join(b).join(c).join(d).primary_key) == [a.c.id]

    def test_join_pk_collapse_explicit(self):
        """test that redundant columns in a join get 'collapsed' into a minimal primary key,
        which is the root column along a chain of explicit join conditions."""

        meta = MetaData()
        a = Table('a', meta, Column('id', Integer, primary_key=True), Column('x', Integer))
        b = Table('b', meta, Column('id', Integer, ForeignKey('a.id'), primary_key=True), Column('x', Integer))
        c = Table('c', meta, Column('id', Integer, ForeignKey('b.id'), primary_key=True), Column('x', Integer))
        d = Table('d', meta, Column('id', Integer, ForeignKey('c.id'), primary_key=True), Column('x', Integer))

        print list(a.join(b, a.c.x==b.c.id).primary_key)
        assert list(a.join(b, a.c.x==b.c.id).primary_key) == [b.c.id]
        assert list(b.join(c, b.c.x==c.c.id).primary_key) == [b.c.id]
        assert list(a.join(b).join(c, c.c.id==b.c.x).primary_key) == [a.c.id]
        assert list(b.join(c, c.c.x==b.c.id).join(d).primary_key) == [c.c.id]
        assert list(b.join(c, c.c.id==b.c.x).join(d).primary_key) == [b.c.id]
        assert list(d.join(b, d.c.id==b.c.id).join(c, b.c.id==c.c.x).primary_key) == [c.c.id]
        assert list(a.join(b).join(c, c.c.id==b.c.x).join(d).primary_key) == [a.c.id]

        assert list(a.join(b, and_(a.c.id==b.c.id, a.c.x==b.c.id)).primary_key) == [a.c.id]

    def test_init_doesnt_blowitaway(self):
        meta = MetaData()
        a = Table('a', meta, Column('id', Integer, primary_key=True), Column('x', Integer))
        b = Table('b', meta, Column('id', Integer, ForeignKey('a.id'), primary_key=True), Column('x', Integer))

        j = a.join(b)
        assert list(j.primary_key) == [a.c.id]

        j.foreign_keys
        assert list(j.primary_key) == [a.c.id]

    def test_non_column_clause(self):
        meta = MetaData()
        a = Table('a', meta, Column('id', Integer, primary_key=True), Column('x', Integer))
        b = Table('b', meta, Column('id', Integer, ForeignKey('a.id'), primary_key=True), Column('x', Integer, primary_key=True))

        j = a.join(b, and_(a.c.id==b.c.id, b.c.x==5))
        assert str(j) == "a JOIN b ON a.id = b.id AND b.x = :b_x_1", str(j)
        assert list(j.primary_key) == [a.c.id, b.c.x]

    def test_onclause_direction(self):
        metadata = MetaData()

        employee = Table( 'Employee', metadata,
            Column('name', String(100)),
            Column('id', Integer, primary_key= True),
        )

        engineer = Table( 'Engineer', metadata,
            Column('id', Integer, ForeignKey( 'Employee.id', ), primary_key=True),
        )

        self.assertEquals(
            set(employee.join(engineer, employee.c.id==engineer.c.id).primary_key),
            set([employee.c.id])
        )

        self.assertEquals(
            set(employee.join(engineer, engineer.c.id==employee.c.id).primary_key),
            set([employee.c.id])
        )


class ReduceTest(TestBase, AssertsExecutionResults):
    def test_reduce(self):
        meta = MetaData()
        t1 = Table('t1', meta,
            Column('t1id', Integer, primary_key=True),
            Column('t1data', String(30)))
        t2 = Table('t2', meta,
            Column('t2id', Integer, ForeignKey('t1.t1id'), primary_key=True),
            Column('t2data', String(30)))
        t3 = Table('t3', meta,
            Column('t3id', Integer, ForeignKey('t2.t2id'), primary_key=True),
            Column('t3data', String(30)))
        
        
        self.assertEquals(
            set(sql_util.reduce_columns([t1.c.t1id, t1.c.t1data, t2.c.t2id, t2.c.t2data, t3.c.t3id, t3.c.t3data])),
            set([t1.c.t1id, t1.c.t1data, t2.c.t2data, t3.c.t3data])
        )
    
    def test_reduce_selectable(self):
       metadata = MetaData()

       engineers = Table('engineers', metadata,
           Column('engineer_id', Integer, primary_key=True),
           Column('engineer_name', String(50)),
          )
     
       managers = Table('managers', metadata,
           Column('manager_id', Integer, primary_key=True),
           Column('manager_name', String(50))
           )

       s = select([engineers, managers]).where(engineers.c.engineer_name==managers.c.manager_name)
       
       self.assertEquals(set(sql_util.reduce_columns(list(s.c), s)),
        set([s.c.engineer_id, s.c.engineer_name, s.c.manager_id])
        )
       
    def test_reduce_aliased_join(self):
        metadata = MetaData()
        people = Table('people', metadata,
           Column('person_id', Integer, Sequence('person_id_seq', optional=True), primary_key=True),
           Column('name', String(50)),
           Column('type', String(30)))

        engineers = Table('engineers', metadata,
           Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
           Column('status', String(30)),
           Column('engineer_name', String(50)),
           Column('primary_language', String(50)),
          )
     
        managers = Table('managers', metadata,
           Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
           Column('status', String(30)),
           Column('manager_name', String(50))
           )
        
        pjoin = people.outerjoin(engineers).outerjoin(managers).select(use_labels=True).alias('pjoin')
        self.assertEquals(
            set(sql_util.reduce_columns([pjoin.c.people_person_id, pjoin.c.engineers_person_id, pjoin.c.managers_person_id])),
            set([pjoin.c.people_person_id])
        )
        
    def test_reduce_aliased_union(self):
        metadata = MetaData()
        item_table = Table(
            'item', metadata,
            Column('id', Integer, ForeignKey('base_item.id'), primary_key=True),
            Column('dummy', Integer, default=0))

        base_item_table = Table(
            'base_item', metadata,
            Column('id', Integer, primary_key=True),
            Column('child_name', String(255), default=None))
        
        from sqlalchemy.orm.util import polymorphic_union
        
        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')
            
        self.assertEquals(
            set(sql_util.reduce_columns([item_join.c.id, item_join.c.dummy, item_join.c.child_name])),
            set([item_join.c.id, item_join.c.dummy, item_join.c.child_name])
        )    
    
    def test_reduce_aliased_union_2(self):
        metadata = MetaData()

        page_table = Table('page', metadata,
            Column('id', Integer, primary_key=True),
        )
        magazine_page_table = Table('magazine_page', metadata,
            Column('page_id', Integer, ForeignKey('page.id'), primary_key=True),
        )
        classified_page_table = Table('classified_page', metadata,
            Column('magazine_page_id', Integer, ForeignKey('magazine_page.page_id'), primary_key=True),
        )
        
        from sqlalchemy.orm.util import polymorphic_union
        pjoin = polymorphic_union(
            {
                'm': page_table.join(magazine_page_table),
                'c': page_table.join(magazine_page_table).join(classified_page_table),
            }, None, 'page_join')
            
        self.assertEquals(
            set(sql_util.reduce_columns([pjoin.c.id, pjoin.c.page_id, pjoin.c.magazine_page_id])),
            set([pjoin.c.id])
        )    
    
            
class DerivedTest(TestBase, AssertsExecutionResults):
    def test_table(self):
        meta = MetaData()
        t1 = Table('t1', meta, Column('c1', Integer, primary_key=True), Column('c2', String(30)))
        t2 = Table('t2', meta, Column('c1', Integer, primary_key=True), Column('c2', String(30)))

        assert t1.is_derived_from(t1)
        assert not t2.is_derived_from(t1)

    def test_alias(self):
        meta = MetaData()
        t1 = Table('t1', meta, Column('c1', Integer, primary_key=True), Column('c2', String(30)))
        t2 = Table('t2', meta, Column('c1', Integer, primary_key=True), Column('c2', String(30)))

        assert t1.alias().is_derived_from(t1)
        assert not t2.alias().is_derived_from(t1)
        assert not t1.is_derived_from(t1.alias())
        assert not t1.is_derived_from(t2.alias())

    def test_select(self):
        meta = MetaData()
        t1 = Table('t1', meta, Column('c1', Integer, primary_key=True), Column('c2', String(30)))
        t2 = Table('t2', meta, Column('c1', Integer, primary_key=True), Column('c2', String(30)))

        assert t1.select().is_derived_from(t1)
        assert not t2.select().is_derived_from(t1)

        assert select([t1, t2]).is_derived_from(t1)

        assert t1.select().alias('foo').is_derived_from(t1)
        assert select([t1, t2]).alias('foo').is_derived_from(t1)
        assert not t2.select().alias('foo').is_derived_from(t1)

if __name__ == "__main__":
    testenv.main()