polymorph2.py

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

        engineer_mapper = mapper(Engineer, engineers, inherits=person_mapper, polymorphic_identity='engineer')
        manager_mapper  = mapper(Manager, managers, inherits=person_mapper, polymorphic_identity='manager')
        car_mapper      = mapper(Car, cars, properties= {'employee':relation(person_mapper)})

        print class_mapper(Person).primary_key
        print person_mapper.get_select_mapper().primary_key

        session = create_session()

        # creating 5 managers named from M1 to E5
        for i in range(1,5):
            session.save(Manager(name="M%d" % i,longer_status="YYYYYYYYY"))
        # creating 5 engineers named from E1 to E5
        for i in range(1,5):
            session.save(Engineer(name="E%d" % i,status="X"))

        session.flush()

        engineer4 = session.query(Engineer).filter(Engineer.name=="E4").first()
        manager3 = session.query(Manager).filter(Manager.name=="M3").first()

        car1 = Car(employee=engineer4)
        session.save(car1)
        car2 = Car(employee=manager3)
        session.save(car2)
        session.flush()

        session.clear()

        print "----------------------------"
        car1 = session.query(Car).get(car1.car_id)
        print "----------------------------"
        usingGet = session.query(person_mapper).get(car1.owner)
        print "----------------------------"
        usingProperty = car1.employee
        print "----------------------------"

        # All print should output the same person (engineer E4)
        assert str(engineer4) == "Engineer E4, status X"
        print str(usingGet)
        assert str(usingGet) == "Engineer E4, status X"
        assert str(usingProperty) == "Engineer E4, status X"

        session.clear()
        print "-----------------------------------------------------------------"
        # and now for the lightning round, eager !

        def go():
            testcar = session.query(Car).options(eagerload('employee')).get(car1.car_id)
            assert str(testcar.employee) == "Engineer E4, status X"
        self.assert_sql_count(testing.db, go, 1)

        session.clear()
        s = session.query(Car)
        c = s.join("employee").filter(Person.name=="E4")[0]
        assert c.car_id==car1.car_id

class RelationTest5(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)),
           Column('type', 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 testeagerempty(self):
        """an easy one...test parent object with child relation to an inheriting mapper, using eager loads,
        works when there are no child objects present"""
        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

        person_mapper   = mapper(Person, people, polymorphic_on=people.c.type, polymorphic_identity='person')
        engineer_mapper = mapper(Engineer, engineers, inherits=person_mapper, polymorphic_identity='engineer')
        manager_mapper  = mapper(Manager, managers, inherits=person_mapper, polymorphic_identity='manager')
        car_mapper      = mapper(Car, cars, properties= {'manager':relation(manager_mapper, lazy=False)})

        sess = create_session()
        car1 = Car()
        car2 = Car()
        car2.manager = Manager()
        sess.save(car1)
        sess.save(car2)
        sess.flush()
        sess.clear()

        carlist = sess.query(Car).all()
        assert carlist[0].manager is None
        assert carlist[1].manager.person_id == car2.manager.person_id

class RelationTest6(ORMTest):
    """test self-referential relationships on a single joined-table inheritance mapper"""
    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)),
           )

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

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

        mapper(Person, people)
        # relationship is from people.join(managers) -> people.join(managers).  self referential logic
        # needs to be used to figure out the lazy clause, meaning create_lazy_clause must go from parent.mapped_table
        # to parent.mapped_table
        mapper(Manager, managers, inherits=Person, inherit_condition=people.c.person_id==managers.c.person_id,
              properties={
                'colleague':relation(Manager, primaryjoin=managers.c.colleague_id==managers.c.person_id, lazy=True, uselist=False)
             }
        )

        sess = create_session()
        m = Manager(name='manager1')
        m2 =Manager(name='manager2')
        m.colleague = m2
        sess.save(m)
        sess.flush()

        sess.clear()
        m = sess.query(Manager).get(m.person_id)
        m2 = sess.query(Manager).get(m2.person_id)
        assert m.colleague is m2

class RelationTest7(ORMTest):
    def define_tables(self, metadata):
        global people, engineers, managers, cars, offroad_cars
        cars = Table('cars', metadata,
                Column('car_id', Integer, primary_key=True),
                Column('name', String(30)))

        offroad_cars = Table('offroad_cars', metadata,
                Column('car_id',Integer, ForeignKey('cars.car_id'),nullable=False,primary_key=True))

        people = Table('people', metadata,
                Column('person_id', Integer, primary_key=True),
                Column('car_id', Integer, ForeignKey('cars.car_id'), nullable=False),
                Column('name', String(50)))

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


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

    def test_manytoone_lazyload(self):
        """test that lazy load clause to a polymorphic child mapper generates correctly [ticket:493]"""
        class PersistentObject(object):
            def __init__(self, **kwargs):
                for key, value in kwargs.iteritems():
                    setattr(self, key, value)

        class Status(PersistentObject):
            def __repr__(self):
                return "Status %s" % self.name

        class Person(PersistentObject):
            def __repr__(self):
                return "Ordinary person %s" % self.name

        class Engineer(Person):
            def __repr__(self):
                return "Engineer %s, field %s" % (self.name, self.field)

        class Manager(Person):
            def __repr__(self):
                return "Manager %s, category %s" % (self.name, self.category)

        class Car(PersistentObject):
            def __repr__(self):
                return "Car number %d, name %s" % i(self.car_id, self.name)

        class Offraod_Car(Car):
            def __repr__(self):
                return "Offroad Car number %d, name %s" % (self.car_id,self.name)

        employee_join = polymorphic_union(
                {
                    'engineer':people.join(engineers),
                    'manager':people.join(managers),
                }, "type", 'employee_join')

        car_join = polymorphic_union(
            {
                'car' : cars.outerjoin(offroad_cars).select(offroad_cars.c.car_id == None, fold_equivalents=True),
                'offroad' : cars.join(offroad_cars)
            }, "type", 'car_join')

        car_mapper  = mapper(Car, cars,
                select_table=car_join,polymorphic_on=car_join.c.type,
                polymorphic_identity='car',
                )
        offroad_car_mapper = mapper(Offraod_Car, offroad_cars, inherits=car_mapper, polymorphic_identity='offroad')
        person_mapper = mapper(Person, people,
                select_table=employee_join,polymorphic_on=employee_join.c.type,
                polymorphic_identity='person',
                properties={
                    'car':relation(car_mapper)
                    })
        engineer_mapper = mapper(Engineer, engineers, inherits=person_mapper, polymorphic_identity='engineer')
        manager_mapper  = mapper(Manager, managers, inherits=person_mapper, polymorphic_identity='manager')

        session = create_session()
        basic_car=Car(name="basic")
        offroad_car=Offraod_Car(name="offroad")

        for i in range(1,4):
            if i%2:
                car=Car()
            else:
                car=Offraod_Car()
            session.save(Manager(name="M%d" % i,category="YYYYYYYYY",car=car))
            session.save(Engineer(name="E%d" % i,field="X",car=car))
            session.flush()
            session.clear()

        r = session.query(Person).all()
        for p in r:
            assert p.car_id == p.car.car_id

class GenerativeTest(TestBase, AssertsExecutionResults):
    def setUpAll(self):
        #  cars---owned by---  people (abstract) --- has a --- status
        #   |                  ^    ^                            |
        #   |                  |    |                            |
        #   |          engineers    managers                     |
        #   |                                                    |
        #   +--------------------------------------- has a ------+

        global metadata, status, people, engineers, managers, cars
        metadata = MetaData(testing.db)
        # table definitions
        status = Table('status', metadata,
           Column('status_id', Integer, primary_key=True),
           Column('name', String(20)))

        people = Table('people', metadata,
           Column('person_id', Integer, primary_key=True),
           Column('status_id', Integer, ForeignKey('status.status_id'), nullable=False),
           Column('name', String(50)))

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

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

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

        metadata.create_all()

    def tearDownAll(self):
        metadata.drop_all()
    def tearDown(self):
        clear_mappers()
        for t in metadata.table_iterator(reverse=True):
            t.delete().execute()

    def testjointo(self):
        # class definitions
        class PersistentObject(object):
            def __init__(self, **kwargs):
                for key, value in kwargs.iteritems():
                    setattr(self, key, value)
        class Status(PersistentObject):
            def __repr__(self):
                return "Status %s" % self.name
        class Person(PersistentObject):
            def __repr__(self):
                return "Ordinary person %s" % self.name
        class Engineer(Person):
            def __repr__(self):
                return "Engineer %s, field %s, status %s" % (self.name, self.field, self.status)
        class Manager(Person):
            def __repr__(self):
                return "Manager %s, category %s, status %s" % (self.name, self.category, self.status)
        class Car(PersistentObject):
            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')

        status_mapper   = mapper(Status, status)
        person_mapper   = mapper(Person, people,
            select_table=employee_join,polymorphic_on=employee_join.c.type,
            polymorphic_identity='person', properties={'status':relation(status_mapper)})
        engineer_mapper = mapper(Engineer, engineers, inherits=person_mapper, polymorphic_identity='engineer')
        manager_mapper  = mapper(Manager, managers, inherits=person_mapper, polymorphic_identity='manager')
        car_mapper      = mapper(Car, cars, properties= {'employee':relation(person_mapper), 'status':relation(status_mapper)})

        session = create_session(echo_uow=False)

        active = Status(name="active")
        dead = Status(name="dead")

        session.save(active)
        session.save(dead)
        session.flush()

        # TODO: we haven't created assertions for all the data combinations created here

        # creating 5 managers named from M1 to M5 and 5 engineers named from E1 to E5
        # M4, M5, E4 and E5 are dead
        for i in range(1,5):
            if i<4:
                st=active