cycles.py

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

        sess = create_session()
        sess.save(o1)
        sess.flush()

        # the bug here is that the dependency sort comes up with T1/T2 in a cycle, but there
        # are no T1/T2 objects to be saved.  therefore no "cyclical subtree" gets generated,
        # and one or the other of T1/T2 gets lost, and processors on T3 dont fire off.
        # the test will then fail because the FK's on T3 are not nullable.
        o3 = T3()
        o3.t1 = o1
        o3.t2 = o1.t2
        sess.save(o3)
        sess.flush()


    def test_reflush_2(self):
        """a variant on test_reflush()"""
        class T1(object):pass
        class T2(object):pass
        class T3(object):pass

        mapper(T1, t1, properties={
            't2':relation(T2, primaryjoin=t1.c.t2id==t2.c.id)
        })
        mapper(T2, t2, properties={
            't1':relation(T1, primaryjoin=t2.c.t1id==t1.c.id)
        })
        mapper(T3, t3, properties={
            't1':relation(T1),
            't2':relation(T2)
        })

        o1 = T1()
        o1.t2 = T2()
        sess = create_session()
        sess.save(o1)
        sess.flush()

        # in this case, T1, T2, and T3 tasks will all be in the cyclical
        # tree normally.  the dependency processors for T3 are part of the
        # 'extradeps' collection so they all get assembled into the tree
        # as well.
        o1a = T1()
        o2a = T2()
        sess.save(o1a)
        sess.save(o2a)
        o3b = T3()
        o3b.t1 = o1a
        o3b.t2 = o2a
        sess.save(o3b)

        o3 = T3()
        o3.t1 = o1
        o3.t2 = o1.t2
        sess.save(o3)
        sess.flush()

class BiDirectionalOneToManyTest(TestBase, AssertsExecutionResults):
    """tests two mappers with a one-to-many relation to each other."""
    def setUpAll(self):
        global t1, t2, metadata
        metadata = MetaData(testing.db)
        t1 = Table('t1', metadata,
            Column('c1', Integer, Sequence('t1c1_id_seq', optional=True), primary_key=True),
            Column('c2', Integer, ForeignKey('t2.c1'))
        )
        t2 = Table('t2', metadata,
            Column('c1', Integer, Sequence('t2c1_id_seq', optional=True), primary_key=True),
            Column('c2', Integer, ForeignKey('t1.c1', use_alter=True, name='t1c1_fk'))
        )
        metadata.create_all()
    def tearDownAll(self):
        metadata.drop_all()
    def tearDown(self):
        clear_mappers()
    def testcycle(self):
        class C1(object):pass
        class C2(object):pass

        m2 = mapper(C2, t2, properties={
            'c1s': relation(C1, primaryjoin=t2.c.c1==t1.c.c2, uselist=True)
        })
        m1 = mapper(C1, t1, properties = {
            'c2s' : relation(C2, primaryjoin=t1.c.c1==t2.c.c2, uselist=True)
        })
        a = C1()
        b = C2()
        c = C1()
        d = C2()
        e = C2()
        f = C2()
        a.c2s.append(b)
        d.c1s.append(c)
        b.c1s.append(c)
        sess = create_session()
        [sess.save(x) for x in [a,b,c,d,e,f]]
        sess.flush()

class BiDirectionalOneToManyTest2(TestBase, AssertsExecutionResults):
    """tests two mappers with a one-to-many relation to each other, with a second one-to-many on one of the mappers"""
    def setUpAll(self):
        global t1, t2, t3, metadata
        metadata = MetaData(testing.db)
        t1 = Table('t1', metadata,
            Column('c1', Integer, Sequence('t1c1_id_seq', optional=True), primary_key=True),
            Column('c2', Integer, ForeignKey('t2.c1')),
        )
        t2 = Table('t2', metadata,
            Column('c1', Integer, Sequence('t2c1_id_seq', optional=True), primary_key=True),
            Column('c2', Integer, ForeignKey('t1.c1', use_alter=True, name='t1c1_fq')),
        )
        t3 = Table('t1_data', metadata,
            Column('c1', Integer, Sequence('t1dc1_id_seq', optional=True), primary_key=True),
            Column('t1id', Integer, ForeignKey('t1.c1')),
            Column('data', String(20)))
        metadata.create_all()

    def tearDown(self):
        clear_mappers()

    def tearDownAll(self):
        metadata.drop_all()

    def testcycle(self):
        class C1(object):pass
        class C2(object):pass
        class C1Data(object):
            def __init__(self, data=None):
                self.data = data

        m2 = mapper(C2, t2, properties={
            'c1s': relation(C1, primaryjoin=t2.c.c1==t1.c.c2, uselist=True)
        })
        m1 = mapper(C1, t1, properties = {
            'c2s' : relation(C2, primaryjoin=t1.c.c1==t2.c.c2, uselist=True),
            'data' : relation(mapper(C1Data, t3))
        })

        a = C1()
        b = C2()
        c = C1()
        d = C2()
        e = C2()
        f = C2()
        a.c2s.append(b)
        d.c1s.append(c)
        b.c1s.append(c)
        a.data.append(C1Data('c1data1'))
        a.data.append(C1Data('c1data2'))
        c.data.append(C1Data('c1data3'))
        sess = create_session()
        [sess.save(x) for x in [a,b,c,d,e,f]]
        sess.flush()

        sess.delete(d)
        sess.delete(c)
        sess.flush()

class OneToManyManyToOneTest(TestBase, AssertsExecutionResults):
    """tests two mappers, one has a one-to-many on the other mapper, the other has a separate many-to-one relationship to the first.
    two tests will have a row for each item that is dependent on the other.  without the "post_update" flag, such relationships
    raise an exception when dependencies are sorted."""
    def setUpAll(self):
        global metadata
        metadata = MetaData(testing.db)
        global person
        global ball
        ball = Table('ball', metadata,
         Column('id', Integer, Sequence('ball_id_seq', optional=True), primary_key=True),
         Column('person_id', Integer, ForeignKey('person.id', use_alter=True, name='fk_person_id')),
         Column('data', String(30))
         )
        person = Table('person', metadata,
         Column('id', Integer, Sequence('person_id_seq', optional=True), primary_key=True),
         Column('favorite_ball_id', Integer, ForeignKey('ball.id')),
         Column('data', String(30))
         )

        metadata.create_all()

    def tearDownAll(self):
        metadata.drop_all()

    def tearDown(self):
        clear_mappers()

    def testcycle(self):
        """this test has a peculiar aspect in that it doesnt create as many dependent
        relationships as the other tests, and revealed a small glitch in the circular dependency sorting."""
        class Person(object):
         pass

        class Ball(object):
         pass

        Ball.mapper = mapper(Ball, ball)
        Person.mapper = mapper(Person, person, properties= dict(
         balls = relation(Ball.mapper, primaryjoin=ball.c.person_id==person.c.id, remote_side=ball.c.person_id),
         favorateBall = relation(Ball.mapper, primaryjoin=person.c.favorite_ball_id==ball.c.id, remote_side=person.c.favorite_ball_id),
         )
        )

        b = Ball()
        p = Person()
        p.balls.append(b)
        sess = create_session()
        sess.save(b)
        sess.save(b)
        sess.flush()

    def testpostupdate_m2o(self):
        """tests a cycle between two rows, with a post_update on the many-to-one"""
        class Person(object):
            def __init__(self, data):
                self.data = data

        class Ball(object):
            def __init__(self, data):
                self.data = data

        Ball.mapper = mapper(Ball, ball)
        Person.mapper = mapper(Person, person, properties= dict(
         balls = relation(Ball.mapper, primaryjoin=ball.c.person_id==person.c.id, remote_side=ball.c.person_id, post_update=False, cascade="all, delete-orphan"),
         favorateBall = relation(Ball.mapper, primaryjoin=person.c.favorite_ball_id==ball.c.id, remote_side=person.c.favorite_ball_id, post_update=True),
         )
        )

        b = Ball('some data')
        p = Person('some data')
        p.balls.append(b)
        p.balls.append(Ball('some data'))
        p.balls.append(Ball('some data'))
        p.balls.append(Ball('some data'))
        p.favorateBall = b
        sess = create_session()
        sess.save(b)
        sess.save(p)

        self.assert_sql(testing.db, lambda: sess.flush(), [
            (
                "INSERT INTO person (favorite_ball_id, data) VALUES (:favorite_ball_id, :data)",
                {'favorite_ball_id': None, 'data':'some data'}
            ),
            (
                "INSERT INTO ball (person_id, data) VALUES (:person_id, :data)",
                lambda ctx:{'person_id':p.id, 'data':'some data'}
            ),
            (
                "INSERT INTO ball (person_id, data) VALUES (:person_id, :data)",
                lambda ctx:{'person_id':p.id, 'data':'some data'}
            ),
            (
                "INSERT INTO ball (person_id, data) VALUES (:person_id, :data)",
                lambda ctx:{'person_id':p.id, 'data':'some data'}
            ),
            (
                "INSERT INTO ball (person_id, data) VALUES (:person_id, :data)",
                lambda ctx:{'person_id':p.id, 'data':'some data'}
            ),
            (
                "UPDATE person SET favorite_ball_id=:favorite_ball_id WHERE person.id = :person_id",
                lambda ctx:{'favorite_ball_id':p.favorateBall.id,'person_id':p.id}
            )
        ],
        with_sequences= [
                (
                    "INSERT INTO person (id, favorite_ball_id, data) VALUES (:id, :favorite_ball_id, :data)",
                    lambda ctx:{'id':ctx.last_inserted_ids()[0], 'favorite_ball_id': None, 'data':'some data'}
                ),
                (
                    "INSERT INTO ball (id, person_id, data) VALUES (:id, :person_id, :data)",
                    lambda ctx:{'id':ctx.last_inserted_ids()[0],'person_id':p.id, 'data':'some data'}
                ),
                (
                    "INSERT INTO ball (id, person_id, data) VALUES (:id, :person_id, :data)",
                    lambda ctx:{'id':ctx.last_inserted_ids()[0],'person_id':p.id, 'data':'some data'}
                ),
                (
                    "INSERT INTO ball (id, person_id, data) VALUES (:id, :person_id, :data)",
                    lambda ctx:{'id':ctx.last_inserted_ids()[0],'person_id':p.id, 'data':'some data'}
                ),
                (
                    "INSERT INTO ball (id, person_id, data) VALUES (:id, :person_id, :data)",
                    lambda ctx:{'id':ctx.last_inserted_ids()[0],'person_id':p.id, 'data':'some data'}
                ),
                # heres the post update
                (
                    "UPDATE person SET favorite_ball_id=:favorite_ball_id WHERE person.id = :person_id",
                    lambda ctx:{'favorite_ball_id':p.favorateBall.id,'person_id':p.id}
                )
            ])
        sess.delete(p)
        self.assert_sql(testing.db, lambda: sess.flush(), [
            # heres the post update (which is a pre-update with deletes)
            (
                "UPDATE person SET favorite_ball_id=:favorite_ball_id WHERE person.id = :person_id",
                lambda ctx:{'person_id': p.id, 'favorite_ball_id': None}
            ),