cascade.py

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

    def testdeleteorphan(self):
        class A(object):
            def __init__(self, data):
                self.data = data
        class B(object):
            def __init__(self, data):
                self.data = data

        mapper(A, a, properties={
            # if no backref here, delete-orphan failed until [ticket:427] was fixed
            'bs':relation(B, secondary=atob, cascade="all, delete-orphan")
        })
        mapper(B, b)

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

        a1.bs.remove(b1)
        sess.flush()
        assert atob.count().scalar() ==0
        assert b.count().scalar() == 0
        assert a.count().scalar() == 1

    def testcascadedelete(self):
        class A(object):
            def __init__(self, data):
                self.data = data
        class B(object):
            def __init__(self, data):
                self.data = data

        mapper(A, a, properties={
            'bs':relation(B, secondary=atob, cascade="all, delete-orphan")
        })
        mapper(B, b)

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

        sess.delete(a1)
        sess.flush()
        assert atob.count().scalar() ==0
        assert b.count().scalar() == 0
        assert a.count().scalar() == 0

class UnsavedOrphansTest(ORMTest):
    """tests regarding pending entities that are orphans"""

    def define_tables(self, metadata):
        global users, addresses, User, Address
        users = Table('users', metadata,
            Column('user_id', Integer, Sequence('user_id_seq', optional=True), primary_key = True),
            Column('user_name', String(40)),
        )

        addresses = Table('email_addresses', metadata,
            Column('address_id', Integer, Sequence('address_id_seq', optional=True), primary_key = True),
            Column('user_id', Integer, ForeignKey(users.c.user_id)),
            Column('email_address', String(40)),
        )
        class User(object):pass
        class Address(object):pass

    def test_pending_orphan(self):
        """test that an entity that never had a parent on a delete-orphan cascade cant be saved."""

        mapper(Address, addresses)
        mapper(User, users, properties=dict(
            addresses=relation(Address, cascade="all,delete-orphan", backref="user")
        ))
        s = create_session()
        a = Address()
        s.save(a)
        try:
            s.flush()
        except exceptions.FlushError, e:
            pass
        assert a.address_id is None, "Error: address should not be persistent"

    def test_delete_new_object(self):
        """test that an entity which is attached then detached from its
        parent with a delete-orphan cascade gets counted as an orphan"""

        mapper(Address, addresses)
        mapper(User, users, properties=dict(
            addresses=relation(Address, cascade="all,delete-orphan", backref="user")
        ))
        s = create_session()

        u = User()
        s.save(u)
        s.flush()
        a = Address()
        assert a not in s.new
        u.addresses.append(a)
        u.addresses.remove(a)
        s.delete(u)
        try:
            s.flush() # (erroneously) causes "a" to be persisted
            assert False
        except exceptions.FlushError:
            assert True
        assert a.address_id is None, "Error: address should not be persistent"


class UnsavedOrphansTest2(ORMTest):
    """same test as UnsavedOrphans only three levels deep"""

    def define_tables(self, meta):
        global orders, items, attributes
        orders = Table('orders', meta,
            Column('id', Integer, Sequence('order_id_seq'), primary_key = True),
            Column('name', VARCHAR(50)),

        )
        items = Table('items', meta,
            Column('id', Integer, Sequence('item_id_seq'), primary_key = True),
            Column('order_id', Integer, ForeignKey(orders.c.id), nullable=False),
            Column('name', VARCHAR(50)),

        )
        attributes = Table('attributes', meta,
            Column('id', Integer, Sequence('attribute_id_seq'), primary_key = True),
            Column('item_id', Integer, ForeignKey(items.c.id), nullable=False),
            Column('name', VARCHAR(50)),

        )

    def testdeletechildwithchild(self):
        """test that an entity which is attached then detached from its
        parent with a delete-orphan cascade gets counted as an orphan, as well
        as its own child instances"""

        class Order(object): pass
        class Item(object): pass
        class Attribute(object): pass

        attrMapper = mapper(Attribute, attributes)
        itemMapper = mapper(Item, items, properties=dict(
            attributes=relation(attrMapper, cascade="all,delete-orphan", backref="item")
        ))
        orderMapper = mapper(Order, orders, properties=dict(
            items=relation(itemMapper, cascade="all,delete-orphan", backref="order")
        ))

        s = create_session( )
        order = Order()
        s.save(order)

        item = Item()
        attr = Attribute()
        item.attributes.append(attr)

        order.items.append(item)
        order.items.remove(item) # item is an orphan, but attr is not so flush() tries to save attr
        try:
            s.flush()
            assert False
        except exceptions.FlushError, e:
            print e
            assert True

        assert item.id is None
        assert attr.id is None

class DoubleParentOrphanTest(TestBase, AssertsExecutionResults):
    """test orphan detection for an entity with two parent relations"""

    def setUpAll(self):
        global metadata, address_table, businesses, homes
        metadata = MetaData(testing.db)
        address_table = Table('addresses', metadata,
            Column('address_id', Integer, primary_key=True),
            Column('street', String(30)),
        )

        homes = Table('homes', metadata,
            Column('home_id', Integer, primary_key=True),
            Column('description', String(30)),
            Column('address_id', Integer, ForeignKey('addresses.address_id'), nullable=False),
        )

        businesses = Table('businesses', metadata,
            Column('business_id', Integer, primary_key=True, key="id"),
            Column('description', String(30), key="description"),
            Column('address_id', Integer, ForeignKey('addresses.address_id'), nullable=False),
        )
        metadata.create_all()
    def tearDown(self):
        clear_mappers()
    def tearDownAll(self):
        metadata.drop_all()
    def test_non_orphan(self):
        """test that an entity can have two parent delete-orphan cascades, and persists normally."""

        class Address(object):pass
        class Home(object):pass
        class Business(object):pass
        mapper(Address, address_table)
        mapper(Home, homes, properties={'address':relation(Address, cascade="all,delete-orphan")})
        mapper(Business, businesses, properties={'address':relation(Address, cascade="all,delete-orphan")})

        session = create_session()
        a1 = Address()
        a2 = Address()
        h1 = Home()
        b1 = Business()
        h1.address = a1
        b1.address = a2
        [session.save(x) for x in [h1,b1]]
        session.flush()

    def test_orphan(self):
        """test that an entity can have two parent delete-orphan cascades, and is detected as an orphan
        when saved without a parent."""

        class Address(object):pass
        class Home(object):pass
        class Business(object):pass
        mapper(Address, address_table)
        mapper(Home, homes, properties={'address':relation(Address, cascade="all,delete-orphan")})
        mapper(Business, businesses, properties={'address':relation(Address, cascade="all,delete-orphan")})

        session = create_session()
        a1 = Address()
        session.save(a1)
        try:
            session.flush()
            assert False
        except exceptions.FlushError, e:
            assert True

class CollectionAssignmentOrphanTest(TestBase, AssertsExecutionResults):
    def setUpAll(self):
        global metadata, table_a, table_b

        metadata = MetaData(testing.db)
        table_a = Table('a', metadata,
                        Column('id', Integer, primary_key=True),
                        Column('foo', String(30)))
        table_b = Table('b', metadata,
                        Column('id', Integer, primary_key=True),
                        Column('foo', String(30)),
                        Column('a_id', Integer, ForeignKey('a.id')))
        metadata.create_all()

    def tearDown(self):
        clear_mappers()
    def tearDownAll(self):
        metadata.drop_all()

    def test_basic(self):
        class A(object):
            def __init__(self, foo):
                self.foo = foo
        class B(object):
            def __init__(self, foo):
                self.foo = foo

        mapper(A, table_a, properties={
            'bs':relation(B, cascade="all, delete-orphan")
            })
        mapper(B, table_b)

        a1 = A('a1')
        a1.bs.append(B('b1'))
        a1.bs.append(B('b2'))
        a1.bs.append(B('b3'))

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

        assert table_b.count(table_b.c.a_id == None).scalar() == 0

        assert table_b.count().scalar() == 3

        a1 = sess.query(A).get(a1.id)
        assert len(a1.bs) == 3
        a1.bs = list(a1.bs)
        assert not class_mapper(B)._is_orphan(a1.bs[0])
        a1.bs[0].foo='b2modified'
        a1.bs[1].foo='b3modified'
        sess.flush()

        assert table_b.count().scalar() == 3

if __name__ == "__main__":
    testenv.main()