cycles.py

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

import testenv; testenv.configure_for_tests()
from sqlalchemy import *
from sqlalchemy.orm import *
from testlib import *
from testlib.tables import *

"""
Tests cyclical mapper relationships.

We might want to try an automated generate of much of this, all combos of
T1<->T2, with o2m or m2o between them, and a third T3 with o2m/m2o to one/both
T1/T2.
"""


class Tester(object):
    def __init__(self, data=None):
        self.data = data
        print repr(self) + " (%d)" % (id(self))
    def __repr__(self):
        return "%s(%s)" % (self.__class__.__name__, repr(self.data))

class SelfReferentialTest(TestBase, AssertsExecutionResults):
    """tests a self-referential mapper, with an additional list of child objects."""
    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('parent_c1', Integer, ForeignKey('t1.c1')),
            Column('data', String(20))
        )
        t2 = Table('t2', metadata,
            Column('c1', Integer, Sequence('t2c1_id_seq', optional=True), primary_key=True),
            Column('c1id', Integer, ForeignKey('t1.c1')),
            Column('data', String(20))
        )
        metadata.create_all()
    def tearDownAll(self):
        metadata.drop_all()
    def setUp(self):
        clear_mappers()

    def testsingle(self):
        class C1(Tester):
            pass
        m1 = mapper(C1, t1, properties = {
            'c1s':relation(C1, cascade="all"),
            'parent':relation(C1, primaryjoin=t1.c.parent_c1==t1.c.c1, remote_side=t1.c.c1, lazy=True, uselist=False)
        })
        a = C1('head c1')
        a.c1s.append(C1('another c1'))
        sess = create_session( )
        sess.save(a)
        sess.flush()
        sess.delete(a)
        sess.flush()

    def testmanytooneonly(self):
        """test that the circular dependency sort can assemble a many-to-one dependency processor
        when only the object on the "many" side is actually in the list of modified objects.
        this requires that the circular sort add the other side of the relation into the UOWTransaction
        so that the dependency operation can be tacked onto it.

        This also affects inheritance relationships since they rely upon circular sort as well.
        """
        class C1(Tester):
            pass
        mapper(C1, t1, properties={
            'parent':relation(C1, primaryjoin=t1.c.parent_c1==t1.c.c1, remote_side=t1.c.c1)
        })
        sess = create_session()
        c1 = C1()
        sess.save(c1)
        sess.flush()
        sess.clear()
        c1 = sess.query(C1).get(c1.c1)
        c2 = C1()
        c2.parent = c1
        sess.save(c2)
        sess.flush()
        assert c2.parent_c1==c1.c1

    def testcycle(self):
        class C1(Tester):
            pass
        class C2(Tester):
            pass

        m1 = mapper(C1, t1, properties = {
            'c1s' : relation(C1, cascade="all"),
            'c2s' : relation(mapper(C2, t2), cascade="all, delete-orphan")
        })

        a = C1('head c1')
        a.c1s.append(C1('child1'))
        a.c1s.append(C1('child2'))
        a.c1s[0].c1s.append(C1('subchild1'))
        a.c1s[0].c1s.append(C1('subchild2'))
        a.c1s[1].c2s.append(C2('child2 data1'))
        a.c1s[1].c2s.append(C2('child2 data2'))
        sess = create_session( )
        sess.save(a)
        sess.flush()

        sess.delete(a)
        sess.flush()

class SelfReferentialNoPKTest(TestBase, AssertsExecutionResults):
    """test self-referential relationship that joins on a column other than the primary key column"""
    def setUpAll(self):
        global table, meta
        meta = MetaData(testing.db)
        table = Table('item', meta,
           Column('id', Integer, primary_key=True),
           Column('uuid', String(32), unique=True, nullable=False),
           Column('parent_uuid', String(32), ForeignKey('item.uuid'), nullable=True),
        )
        meta.create_all()
    def tearDown(self):
        table.delete().execute()
    def tearDownAll(self):
        meta.drop_all()
    def testbasic(self):
        class TT(object):
            def __init__(self):
                self.uuid = hex(id(self))
        mapper(TT, table, properties={'children':relation(TT, remote_side=[table.c.parent_uuid], backref=backref('parent', remote_side=[table.c.uuid]))})
        s = create_session()
        t1 = TT()
        t1.children.append(TT())
        t1.children.append(TT())
        s.save(t1)
        s.flush()
        s.clear()
        t = s.query(TT).filter_by(id=t1.id).one()
        assert t.children[0].parent_uuid == t1.uuid
    def testlazyclause(self):
        class TT(object):
            def __init__(self):
                self.uuid = hex(id(self))
        mapper(TT, table, properties={'children':relation(TT, remote_side=[table.c.parent_uuid], backref=backref('parent', remote_side=[table.c.uuid]))})
        s = create_session()
        t1 = TT()
        t2 = TT()
        t1.children.append(t2)
        s.save(t1)
        s.flush()
        s.clear()

        t = s.query(TT).filter_by(id=t2.id).one()
        assert t.uuid == t2.uuid
        assert t.parent.uuid == t1.uuid

class InheritTestOne(TestBase, AssertsExecutionResults):
    def setUpAll(self):
        global parent, child1, child2, meta
        meta = MetaData(testing.db)
        parent = Table("parent", meta,
            Column("id", Integer, primary_key=True),
            Column("parent_data", String(50)),
            Column("type", String(10))
            )

        child1 = Table("child1", meta,
            Column("id", Integer, ForeignKey("parent.id"), primary_key=True),
            Column("child1_data", String(50))
            )

        child2 = Table("child2", meta,
            Column("id", Integer, ForeignKey("parent.id"), primary_key=True),
            Column("child1_id", Integer, ForeignKey("child1.id"), nullable=False),
            Column("child2_data", String(50))
            )
        meta.create_all()
    def tearDownAll(self):
        meta.drop_all()
    def testmanytooneonly(self):
        """test similar to SelfReferentialTest.testmanytooneonly"""
        class Parent(object):
                pass

        mapper(Parent, parent)

        class Child1(Parent):
                pass

        mapper(Child1, child1, inherits=Parent)

        class Child2(Parent):
                pass

        mapper(Child2, child2, properties={
                        "child1": relation(Child1,
                                primaryjoin=child2.c.child1_id==child1.c.id,
                        )
                },inherits=Parent)

        session = create_session()

        c1 = Child1()
        c1.child1_data = "qwerty"
        session.save(c1)
        session.flush()
        session.clear()

        c1 = session.query(Child1).filter_by(child1_data="qwerty").one()
        c2 = Child2()
        c2.child1 = c1
        c2.child2_data = "asdfgh"
        session.save(c2)
        # the flush will fail if the UOW does not set up a many-to-one DP
        # attached to a task corresponding to c1, since "child1_id" is not nullable
        session.flush()

class InheritTestTwo(ORMTest):
    """the fix in BiDirectionalManyToOneTest raised this issue, regarding
    the 'circular sort' containing UOWTasks that were still polymorphic, which could
    create duplicate entries in the final sort"""
    def define_tables(self, metadata):
        global a, b, c
        a = Table('a', metadata,
            Column('id', Integer, primary_key=True),
            Column('data', String(30)),
            Column('cid', Integer, ForeignKey('c.id')),
            )

        b = Table('b', metadata,
            Column('id', Integer, ForeignKey("a.id"), primary_key=True),
            Column('data', String(30)),
            )

        c = Table('c', metadata,
            Column('id', Integer, primary_key=True),
            Column('data', String(30)),
            Column('aid', Integer, ForeignKey('a.id', use_alter=True, name="foo")),
            )
    def test_flush(self):
        class A(object):pass
        class B(A):pass
        class C(object):pass

        mapper(A, a, properties={
            'cs':relation(C, primaryjoin=a.c.cid==c.c.id)
        })

        mapper(B, b, inherits=A, inherit_condition=b.c.id==a.c.id, properties={
        })
        mapper(C, c, properties={
            'arel':relation(A, primaryjoin=a.c.id==c.c.aid)
        })

        sess = create_session()
        bobj = B()
        sess.save(bobj)
        cobj = C()
        sess.save(cobj)
        sess.flush()


class BiDirectionalManyToOneTest(ORMTest):
    def define_tables(self, metadata):
        global t1, t2, t3, t4
        t1 = Table('t1', metadata,
            Column('id', Integer, primary_key=True),
            Column('data', String(30)),
            Column('t2id', Integer, ForeignKey('t2.id'))
            )
        t2 = Table('t2', metadata,
            Column('id', Integer, primary_key=True),
            Column('data', String(30)),
            Column('t1id', Integer, ForeignKey('t1.id', use_alter=True, name="foo_fk"))
            )
        t3 = Table('t3', metadata,
            Column('id', Integer, primary_key=True),
            Column('data', String(30)),
            Column('t1id', Integer, ForeignKey('t1.id'), nullable=False),
            Column('t2id', Integer, ForeignKey('t2.id'), nullable=False),
            )

    def test_reflush(self):
        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()