polymorph2.py

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

import testenv; testenv.configure_for_tests()
from sqlalchemy import *
from sqlalchemy import exceptions, util
from sqlalchemy.orm import *
from testlib import *
from testlib import fixtures

class AttrSettable(object):
    def __init__(self, **kwargs):
        [setattr(self, k, v) for k, v in kwargs.iteritems()]
    def __repr__(self):
        return self.__class__.__name__ + "(%s)" % (hex(id(self)))


class RelationTest1(ORMTest):
    """test self-referential relationships on polymorphic mappers"""
    def define_tables(self, metadata):
        global people, managers

        people = Table('people', metadata,
           Column('person_id', Integer, Sequence('person_id_seq', optional=True), primary_key=True),
           Column('manager_id', Integer, ForeignKey('managers.person_id', use_alter=True, name="mpid_fq")),
           Column('name', String(50)),
           Column('type', String(30)))

        managers = Table('managers', metadata,
           Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
           Column('status', String(30)),
           Column('manager_name', String(50))
           )

    def tearDown(self):
        people.update(values={people.c.manager_id:None}).execute()
        super(RelationTest1, self).tearDown()

    def testparentrefsdescendant(self):
        class Person(AttrSettable):
            pass
        class Manager(Person):
            pass

        mapper(Person, people, properties={
            'manager':relation(Manager, primaryjoin=people.c.manager_id==managers.c.person_id, uselist=False)
        })
        mapper(Manager, managers, inherits=Person, inherit_condition=people.c.person_id==managers.c.person_id)
        try:
            compile_mappers()
        except exceptions.ArgumentError, ar:
            assert str(ar) == "Can't determine relation direction for relationship 'Person.manager (Manager)' - foreign key columns are present in both the parent and the child's mapped tables.  Specify 'foreign_keys' argument.", str(ar)

        clear_mappers()

        mapper(Person, people, properties={
            'manager':relation(Manager, primaryjoin=(people.c.manager_id ==
                                                     managers.c.person_id),
                               foreign_keys=[people.c.manager_id],
                               uselist=False, post_update=True)
        })
        mapper(Manager, managers, inherits=Person,
               inherit_condition=people.c.person_id==managers.c.person_id)

        session = create_session()
        p = Person(name='some person')
        m = Manager(name='some manager')
        p.manager = m
        session.save(p)
        session.flush()
        session.clear()

        p = session.query(Person).get(p.person_id)
        m = session.query(Manager).get(m.person_id)
        print p, m, p.manager
        assert p.manager is m

    def testdescendantrefsparent(self):
        class Person(AttrSettable):
            pass
        class Manager(Person):
            pass

        mapper(Person, people)
        mapper(Manager, managers, inherits=Person, inherit_condition=people.c.person_id==managers.c.person_id, properties={
            'employee':relation(Person, primaryjoin=(people.c.manager_id ==
                                                     managers.c.person_id),
                                foreign_keys=[people.c.manager_id],
                                uselist=False, post_update=True)
        })

        session = create_session()
        p = Person(name='some person')
        m = Manager(name='some manager')
        m.employee = p
        session.save(m)
        session.flush()
        session.clear()

        p = session.query(Person).get(p.person_id)
        m = session.query(Manager).get(m.person_id)
        print p, m, m.employee
        assert m.employee is p

class RelationTest2(ORMTest):
    """test self-referential relationships on polymorphic mappers"""
    def define_tables(self, metadata):
        global people, managers, data
        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)))

        managers = Table('managers', metadata,
           Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
           Column('manager_id', Integer, ForeignKey('people.person_id')),
           Column('status', String(30)),
           )

        data = Table('data', metadata,
            Column('person_id', Integer, ForeignKey('managers.person_id'), primary_key=True),
            Column('data', String(30))
            )

    def testrelationonsubclass_j1_nodata(self):
        self.do_test("join1", False)
    def testrelationonsubclass_j2_nodata(self):
        self.do_test("join2", False)
    def testrelationonsubclass_j1_data(self):
        self.do_test("join1", True)
    def testrelationonsubclass_j2_data(self):
        self.do_test("join2", True)
    def testrelationonsubclass_j3_nodata(self):
        self.do_test("join3", False)
    def testrelationonsubclass_j3_data(self):
        self.do_test("join3", True)

    def do_test(self, jointype="join1", usedata=False):
        class Person(AttrSettable):
            pass
        class Manager(Person):
            pass

        if jointype == "join1":
            poly_union = polymorphic_union({
                'person':people.select(people.c.type=='person'),
                'manager':join(people, managers, people.c.person_id==managers.c.person_id)
            }, None)
            polymorphic_on=poly_union.c.type
        elif jointype == "join2":
            poly_union = polymorphic_union({
                'person':people.select(people.c.type=='person'),
                'manager':managers.join(people, people.c.person_id==managers.c.person_id)
            }, None)
            polymorphic_on=poly_union.c.type
        elif jointype == "join3":
            poly_union = None
            polymorphic_on = people.c.type

        if usedata:
            class Data(object):
                def __init__(self, data):
                    self.data = data
            mapper(Data, data)

        mapper(Person, people, select_table=poly_union, polymorphic_identity='person', polymorphic_on=polymorphic_on)

        if usedata:
            mapper(Manager, managers, inherits=Person, inherit_condition=people.c.person_id==managers.c.person_id, polymorphic_identity='manager',
                  properties={
                    'colleague':relation(Person, primaryjoin=managers.c.manager_id==people.c.person_id, lazy=True, uselist=False),
                    'data':relation(Data, uselist=False)
                 }
            )
        else:
            mapper(Manager, managers, inherits=Person, inherit_condition=people.c.person_id==managers.c.person_id, polymorphic_identity='manager',
                  properties={
                    'colleague':relation(Person, primaryjoin=managers.c.manager_id==people.c.person_id, lazy=True, uselist=False)
                 }
            )

        sess = create_session()
        p = Person(name='person1')
        m = Manager(name='manager1')
        m.colleague = p
        if usedata:
            m.data = Data('ms data')
        sess.save(m)
        sess.flush()

        sess.clear()
        p = sess.query(Person).get(p.person_id)
        m = sess.query(Manager).get(m.person_id)
        print p
        print m
        assert m.colleague is p
        if usedata:
            assert m.data.data == 'ms data'

class RelationTest3(ORMTest):
    """test self-referential relationships on polymorphic mappers"""
    def define_tables(self, metadata):
        global people, managers, data
        people = Table('people', metadata,
           Column('person_id', Integer, Sequence('person_id_seq', optional=True), primary_key=True),
           Column('colleague_id', Integer, ForeignKey('people.person_id')),
           Column('name', String(50)),
           Column('type', String(30)))

        managers = Table('managers', metadata,
           Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
           Column('status', String(30)),
           )

        data = Table('data', metadata,
           Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
           Column('data', String(30))
           )

def generate_test(jointype="join1", usedata=False):
    def do_test(self):
        class Person(AttrSettable):
            pass
        class Manager(Person):
            pass

        if usedata:
            class Data(object):
                def __init__(self, data):
                    self.data = data

        if jointype == "join1":
            poly_union = polymorphic_union({
                'manager':managers.join(people, people.c.person_id==managers.c.person_id),
                'person':people.select(people.c.type=='person')
            }, None)
        elif jointype =="join2":
            poly_union = polymorphic_union({
                'manager':join(people, managers, people.c.person_id==managers.c.person_id),
                'person':people.select(people.c.type=='person')
            }, None)
        elif jointype == 'join3':
            poly_union = people.outerjoin(managers)
        elif jointype == "join4":
            poly_union=None

        if usedata:
            mapper(Data, data)

        if usedata:
            mapper(Person, people, select_table=poly_union, polymorphic_identity='person', polymorphic_on=people.c.type,
                  properties={
                    'colleagues':relation(Person, primaryjoin=people.c.colleague_id==people.c.person_id, remote_side=people.c.colleague_id, uselist=True),
                    'data':relation(Data, uselist=False)
                    }
            )
        else:
            mapper(Person, people, select_table=poly_union, polymorphic_identity='person', polymorphic_on=people.c.type,
                  properties={
                    'colleagues':relation(Person, primaryjoin=people.c.colleague_id==people.c.person_id,
                        remote_side=people.c.colleague_id, uselist=True)
                    }
            )

        mapper(Manager, managers, inherits=Person, inherit_condition=people.c.person_id==managers.c.person_id, polymorphic_identity='manager')

        sess = create_session()
        p = Person(name='person1')
        p2 = Person(name='person2')
        p3 = Person(name='person3')
        m = Manager(name='manager1')
        p.colleagues.append(p2)
        m.colleagues.append(p3)
        if usedata:
            p.data = Data('ps data')
            m.data = Data('ms data')

        sess.save(m)
        sess.save(p)
        sess.flush()

        sess.clear()
        p = sess.query(Person).get(p.person_id)
        p2 = sess.query(Person).get(p2.person_id)
        p3 = sess.query(Person).get(p3.person_id)
        m = sess.query(Person).get(m.person_id)
        print p, p2, p.colleagues, m.colleagues
        assert len(p.colleagues) == 1
        assert p.colleagues == [p2]
        assert m.colleagues == [p3]
        if usedata:
            assert p.data.data == 'ps data'
            assert m.data.data == 'ms data'

    do_test = _function_named(
        do_test, 'test_relationonbaseclass_%s_%s' % (
        jointype, data and "nodata" or "data"))
    return do_test

for jointype in ["join1", "join2", "join3", "join4"]:
    for data in (True, False):
        func = generate_test(jointype, data)
        setattr(RelationTest3, func.__name__, func)


class RelationTest4(ORMTest):
    def define_tables(self, metadata):
        global people, engineers, managers, cars
        people = Table('people', metadata,
           Column('person_id', Integer, primary_key=True),
           Column('name', String(50)))

        engineers = Table('engineers', metadata,
           Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
           Column('status', String(30)))

        managers = Table('managers', metadata,
           Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
           Column('longer_status', String(70)))

        cars = Table('cars', metadata,
           Column('car_id', Integer, primary_key=True),
           Column('owner', Integer, ForeignKey('people.person_id')))

    def testmanytoonepolymorphic(self):
        """in this test, the polymorphic union is between two subclasses, but does not include the base table by itself
         in the union.  however, the primaryjoin condition is going to be against the base table, and its a many-to-one
         relationship (unlike the test in polymorph.py) so the column in the base table is explicit.  Can the ClauseAdapter
         figure out how to alias the primaryjoin to the polymorphic union ?"""

        # class definitions
        class Person(object):
            def __init__(self, **kwargs):
                for key, value in kwargs.iteritems():
                    setattr(self, key, value)
            def __repr__(self):
                return "Ordinary person %s" % self.name
        class Engineer(Person):
            def __repr__(self):
                return "Engineer %s, status %s" % (self.name, self.status)
        class Manager(Person):
            def __repr__(self):
                return "Manager %s, status %s" % (self.name, self.longer_status)
        class Car(object):
            def __init__(self, **kwargs):
                for key, value in kwargs.iteritems():
                    setattr(self, key, value)
            def __repr__(self):
                return "Car number %d" % self.car_id

        # create a union that represents both types of joins.
        employee_join = polymorphic_union(
            {
                'engineer':people.join(engineers),
                'manager':people.join(managers),
            }, "type", 'employee_join')

        person_mapper   = mapper(Person, people, select_table=employee_join,polymorphic_on=employee_join.c.type, polymorphic_identity='person')