relationships.py

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

        a1 = A()
        a1.bs.append(B())
        sess = create_session()
        sess.save(a1)
        sess.flush()

        sess.delete(a1)
        try:
            sess.flush()
            assert False
        except exceptions.AssertionError, e:
            assert str(e).startswith("Dependency rule tried to blank-out primary key column 'B.id' on instance ")

    def test_no_delete_PK_BtoA(self):
        class A(object):pass
        class B(object):pass
        mapper(B, tableB, properties={
            'a':relation(A, cascade="save-update")
        })
        mapper(A, tableA)
        b1 = B()
        a1 = A()
        b1.a = a1
        sess = create_session()
        sess.save(b1)
        sess.flush()
        b1.a = None
        try:
            sess.flush()
            assert False
        except exceptions.AssertionError, e:
            assert str(e).startswith("Dependency rule tried to blank-out primary key column 'B.id' on instance ")

    @testing.fails_on_everything_except('sqlite', 'mysql')
    def test_nullPKsOK_BtoA(self):
        # postgres cant handle a nullable PK column...?
        tableC = Table('tablec', tableA.metadata,
            Column('id', Integer, primary_key=True),
            Column('a_id', Integer, ForeignKey('A.id'), primary_key=True, autoincrement=False, nullable=True))
        tableC.create()

        class A(object):pass
        class C(object):pass
        mapper(C, tableC, properties={
            'a':relation(A, cascade="save-update")
        }, allow_null_pks=True)
        mapper(A, tableA)
        c1 = C()
        c1.id = 5
        c1.a = None
        sess = create_session()
        sess.save(c1)
        # test that no error is raised.
        sess.flush()

    def test_delete_cascade_BtoA(self):
        """test that the 'blank the PK' error doesnt get raised when the child is to be deleted as part of a
        cascade"""
        class A(object):pass
        class B(object):pass
        for cascade in (
                    "save-update, delete",
                    #"save-update, delete-orphan",
                    "save-update, delete, delete-orphan"):

            mapper(B, tableB, properties={
                'a':relation(A, cascade=cascade)
            })
            mapper(A, tableA)
            b1 = B()
            a1 = A()
            b1.a = a1
            sess = create_session()
            sess.save(b1)
            sess.flush()
            sess.delete(b1)
            sess.flush()
            assert a1 not in sess
            assert b1 not in sess
            sess.clear()
            clear_mappers()

    def test_delete_cascade_AtoB(self):
        """test that the 'blank the PK' error doesnt get raised when the child is to be deleted as part of a
        cascade"""
        class A(object):pass
        class B(object):pass
        for cascade in (
                    "save-update, delete",
                    #"save-update, delete-orphan",
                    "save-update, delete, delete-orphan"):
            mapper(A, tableA, properties={
                'bs':relation(B, cascade=cascade)
            })
            mapper(B, tableB)
            a1 = A()
            b1 = B()
            a1.bs.append(b1)
            sess = create_session()
            sess.save(a1)
            sess.flush()

            sess.delete(a1)
            sess.flush()
            assert a1 not in sess
            assert b1 not in sess
            sess.clear()
            clear_mappers()

    def test_delete_manual_AtoB(self):
        class A(object):pass
        class B(object):pass
        mapper(A, tableA, properties={
            'bs':relation(B, cascade="none")
        })
        mapper(B, tableB)
        a1 = A()
        b1 = B()
        a1.bs.append(b1)
        sess = create_session()
        sess.save(a1)
        sess.save(b1)
        sess.flush()

        sess.delete(a1)
        sess.delete(b1)
        sess.flush()
        assert a1 not in sess
        assert b1 not in sess
        sess.clear()

    def test_delete_manual_BtoA(self):
        class A(object):pass
        class B(object):pass
        mapper(B, tableB, properties={
            'a':relation(A, cascade="none")
        })
        mapper(A, tableA)
        b1 = B()
        a1 = A()
        b1.a = a1
        sess = create_session()
        sess.save(b1)
        sess.save(a1)
        sess.flush()
        sess.delete(b1)
        sess.delete(a1)
        sess.flush()
        assert a1 not in sess
        assert b1 not in sess

class RelationTest5(ORMTest):
    """Test a map to a select that relates to a map to the table."""

    def define_tables(self, metadata):
        global items
        items = Table('items', metadata,
            Column('item_policy_num', String(10), primary_key=True, key='policyNum'),
            Column('item_policy_eff_date', Date, primary_key=True, key='policyEffDate'),
            Column('item_type', String(20), primary_key=True, key='type'),
            Column('item_id', Integer, primary_key=True, key='id', autoincrement=False),
        )

    def test_basic(self):
        class Container(object):pass
        class LineItem(object):pass

        container_select = select(
            [items.c.policyNum, items.c.policyEffDate, items.c.type],
            distinct=True,
            ).alias('container_select')

        mapper(LineItem, items)

        mapper(Container, container_select, order_by=asc(container_select.c.type), properties=dict(
            lineItems = relation(LineItem, lazy=True, cascade='all, delete-orphan', order_by=asc(items.c.type),
                primaryjoin=and_(
                    container_select.c.policyNum==items.c.policyNum,
                    container_select.c.policyEffDate==items.c.policyEffDate,
                    container_select.c.type==items.c.type
                ),
                foreign_keys=[
                    items.c.policyNum,
                    items.c.policyEffDate,
                    items.c.type,
                ],
            )
        ))
        session = create_session()
        con = Container()
        con.policyNum = "99"
        con.policyEffDate = datetime.date.today()
        con.type = "TESTER"
        session.save(con)
        for i in range(0, 10):
            li = LineItem()
            li.id = i
            con.lineItems.append(li)
            session.save(li)
        session.flush()
        session.clear()
        newcon = session.query(Container).first()
        assert con.policyNum == newcon.policyNum
        assert len(newcon.lineItems) == 10
        for old, new in zip(con.lineItems, newcon.lineItems):
            assert old.id == new.id


class TypeMatchTest(ORMTest):
    """test errors raised when trying to add items whose type is not handled by a relation"""
    def define_tables(self, metadata):
        global a, b, c, d
        a = Table("a", metadata,
            Column('aid', Integer, primary_key=True),
            Column('data', String(30)))
        b = Table("b", metadata,
            Column('bid', Integer, primary_key=True),
            Column("a_id", Integer, ForeignKey("a.aid")),
            Column('data', String(30)))
        c = Table("c", metadata,
            Column('cid', Integer, primary_key=True),
            Column("b_id", Integer, ForeignKey("b.bid")),
            Column('data', String(30)))
        d = Table("d", metadata,
            Column('did', Integer, primary_key=True),
            Column("a_id", Integer, ForeignKey("a.aid")),
            Column('data', String(30)))
    def test_o2m_oncascade(self):
        class A(object):pass
        class B(object):pass
        class C(object):pass
        mapper(A, a, properties={'bs':relation(B)})
        mapper(B, b)
        mapper(C, c)

        a1 = A()
        b1 = B()
        c1 = C()
        a1.bs.append(b1)
        a1.bs.append(c1)
        sess = create_session()
        try:
            sess.save(a1)
            assert False
        except exceptions.AssertionError, err:
            assert str(err) == "Attribute 'bs' on class '%s' doesn't handle objects of type '%s'" % (A, C)
    def test_o2m_onflush(self):
        class A(object):pass
        class B(object):pass
        class C(object):pass
        mapper(A, a, properties={'bs':relation(B, cascade="none")})
        mapper(B, b)
        mapper(C, c)

        a1 = A()
        b1 = B()
        c1 = C()
        a1.bs.append(b1)
        a1.bs.append(c1)
        sess = create_session()
        sess.save(a1)
        sess.save(b1)
        sess.save(c1)
        try:
            sess.flush()
            assert False
        except exceptions.FlushError, err:
            assert str(err).startswith("Attempting to flush an item of type %s on collection 'A.bs (B)', which is handled by mapper 'Mapper|B|b' and does not load items of that type.  Did you mean to use a polymorphic mapper for this relationship ?" % C)
    def test_o2m_nopoly_onflush(self):
        class A(object):pass
        class B(object):pass
        class C(B):pass
        mapper(A, a, properties={'bs':relation(B, cascade="none")})
        mapper(B, b)
        mapper(C, c, inherits=B)

        a1 = A()
        b1 = B()
        c1 = C()
        a1.bs.append(b1)
        a1.bs.append(c1)
        sess = create_session()
        sess.save(a1)
        sess.save(b1)
        sess.save(c1)
        try:
            sess.flush()
            assert False
        except exceptions.FlushError, err:
            assert str(err).startswith("Attempting to flush an item of type %s on collection 'A.bs (B)', which is handled by mapper 'Mapper|B|b' and does not load items of that type.  Did you mean to use a polymorphic mapper for this relationship ?" % C)

    def test_m2o_nopoly_onflush(self):
        class A(object):pass
        class B(A):pass
        class D(object):pass
        mapper(A, a)
        mapper(B, b, inherits=A)
        mapper(D, d, properties={"a":relation(A, cascade="none")})
        b1 = B()
        d1 = D()
        d1.a = b1
        sess = create_session()
        sess.save(b1)
        sess.save(d1)
        try:
            sess.flush()
            assert False
        except exceptions.FlushError, err:
            assert str(err).startswith("Attempting to flush an item of type %s on collection 'D.a (A)', which is handled by mapper 'Mapper|A|a' and does not load items of that type.  Did you mean to use a polymorphic mapper for this relationship ?" % B)
    def test_m2o_oncascade(self):
        class A(object):pass
        class B(object):pass
        class D(object):pass
        mapper(A, a)
        mapper(B, b)
        mapper(D, d, properties={"a":relation(A)})
        b1 = B()
        d1 = D()
        d1.a = b1
        sess = create_session()
        try:
            sess.save(d1)
            assert False
        except exceptions.AssertionError, err:
            assert str(err) == "Attribute 'a' on class '%s' doesn't handle objects of type '%s'" % (D, B)

class TypedAssociationTable(ORMTest):
    def define_tables(self, metadata):
        global t1, t2, t3

        class MySpecialType(types.TypeDecorator):
            impl = String
            def convert_bind_param(self, value, dialect):
                return "lala" + value
            def convert_result_value(self, value, dialect):
                return value[4:]

        t1 = Table('t1', metadata,
            Column('col1', MySpecialType(30), primary_key=True),
            Column('col2', String(30)))
        t2 = Table('t2', metadata,
            Column('col1', MySpecialType(30), primary_key=True),
            Column('col2', String(30)))
        t3 = Table('t3', metadata,
            Column('t1c1', MySpecialType(30), ForeignKey('t1.col1')),
            Column('t2c1', MySpecialType(30), ForeignKey('t2.col1')),
        )
    def testm2m(self):
        """test many-to-many tables with special types for candidate keys"""

        class T1(object):pass
        class T2(object):pass
        mapper(T2, t2)
        mapper(T1, t1, properties={
            't2s':relation(T2, secondary=t3, backref='t1s')
        })
        a = T1()
        a.col1 = "aid"
        b = T2()
        b.col1 = "bid"
        c = T2()
        c.col1 = "cid"
        a.t2s.append(b)
        a.t2s.append(c)
        sess = create_session()
        sess.save(a)
        sess.flush()

        assert t3.count().scalar() == 2

        a.t2s.remove(c)
        sess.flush()

        assert t3.count().scalar() == 1

# TODO: move these tests to either attributes.py test or its own module
class CustomCollectionsTest(ORMTest):
    def define_tables(self, metadata):
        global sometable, someothertable
        sometable = Table('sometable', metadata,
            Column('col1',Integer, primary_key=True),
            Column('data', String(30)))
        someothertable = Table('someothertable', metadata,
            Column('col1', Integer, primary_key=True),