cycles.py

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

            (
                "DELETE FROM ball WHERE ball.id = :id",
                None
                # order cant be predicted, but something like:
                #lambda ctx:[{'id': 1L}, {'id': 4L}, {'id': 3L}, {'id': 2L}]
            ),
            (
                "DELETE FROM person WHERE person.id = :id",
                lambda ctx:[{'id': p.id}]
            )


        ])

    def testpostupdate_o2m(self):
        """tests a cycle between two rows, with a post_update on the one-to-many"""
        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, cascade="all, delete-orphan", post_update=True, backref='person'),
         favorateBall = relation(Ball.mapper, primaryjoin=person.c.favorite_ball_id==ball.c.id, remote_side=person.c.favorite_ball_id),
         )
        )

        b = Ball('some data')
        p = Person('some data')
        p.balls.append(b)
        b2 = Ball('some data')
        p.balls.append(b2)
        b3 = Ball('some data')
        p.balls.append(b3)
        b4 = Ball('some data')
        p.balls.append(b4)
        p.favorateBall = b
        sess = create_session()
        [sess.save(x) for x in [b,p,b2,b3,b4]]

        self.assert_sql(testing.db, lambda: sess.flush(), [
                (
                    "INSERT INTO ball (person_id, data) VALUES (:person_id, :data)",
                    {'person_id':None, 'data':'some data'}
                ),
                (
                    "INSERT INTO ball (person_id, data) VALUES (:person_id, :data)",
                    {'person_id':None, 'data':'some data'}
                ),
                (
                    "INSERT INTO ball (person_id, data) VALUES (:person_id, :data)",
                    {'person_id':None, 'data':'some data'}
                ),
                (
                    "INSERT INTO ball (person_id, data) VALUES (:person_id, :data)",
                    {'person_id':None, 'data':'some data'}
                ),
                (
                    "INSERT INTO person (favorite_ball_id, data) VALUES (:favorite_ball_id, :data)",
                    lambda ctx:{'favorite_ball_id':b.id, 'data':'some data'}
                ),
                # heres the post update on each one-to-many item
                (
                    "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
                    lambda ctx:{'person_id':p.id,'ball_id':b.id}
                ),
                (
                    "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
                    lambda ctx:{'person_id':p.id,'ball_id':b2.id}
                ),
                (
                    "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
                    lambda ctx:{'person_id':p.id,'ball_id':b3.id}
                ),
                (
                    "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
                    lambda ctx:{'person_id':p.id,'ball_id':b4.id}
                ),
        ],
        with_sequences=[
            (
                "INSERT INTO ball (id, person_id, data) VALUES (:id, :person_id, :data)",
                lambda ctx:{'id':ctx.last_inserted_ids()[0], 'person_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':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':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':None, 'data':'some data'}
            ),
            (
                "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':b.id, 'data':'some data'}
            ),
            (
                "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
                lambda ctx:{'person_id':p.id,'ball_id':b.id}
            ),
            (
                "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
                lambda ctx:{'person_id':p.id,'ball_id':b2.id}
            ),
            (
                "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
                lambda ctx:{'person_id':p.id,'ball_id':b3.id}
            ),
            (
                "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
                lambda ctx:{'person_id':p.id,'ball_id':b4.id}
            ),
        ])

        sess.delete(p)
        self.assert_sql(testing.db, lambda: sess.flush(), [
            (
                "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
                lambda ctx:{'person_id': None, 'ball_id': b.id}
            ),
            (
                "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
                lambda ctx:{'person_id': None, 'ball_id': b2.id}
            ),
            (
                "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
                lambda ctx:{'person_id': None, 'ball_id': b3.id}
            ),
            (
                "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
                lambda ctx:{'person_id': None, 'ball_id': b4.id}
            ),
            (
                "DELETE FROM person WHERE person.id = :id",
                lambda ctx:[{'id':p.id}]
            ),
            (
                "DELETE FROM ball WHERE ball.id = :id",
                lambda ctx:[{'id': b.id}, {'id': b2.id}, {'id': b3.id}, {'id': b4.id}]
            )
        ])

class SelfReferentialPostUpdateTest(TestBase, AssertsExecutionResults):
    """test using post_update on a single self-referential mapper"""
    def setUpAll(self):
        global metadata, node_table
        metadata = MetaData(testing.db)
        node_table = Table('node', metadata,
            Column('id', Integer, Sequence('nodeid_id_seq', optional=True), primary_key=True),
            Column('path', String(50), nullable=False),
            Column('parent_id', Integer, ForeignKey('node.id'), nullable=True),
            Column('prev_sibling_id', Integer, ForeignKey('node.id'), nullable=True),
            Column('next_sibling_id', Integer, ForeignKey('node.id'), nullable=True)
        )
        node_table.create()
    def tearDownAll(self):
        node_table.drop()

    def testbasic(self):
        """test that post_update only fires off when needed.

        this test case used to produce many superfluous update statements, particularly upon delete"""
        class Node(object):
            def __init__(self, path=''):
                self.path = path

        n_mapper = mapper(Node, node_table, properties={
            'children': relation(
                Node,
                primaryjoin=node_table.c.id==node_table.c.parent_id,
                lazy=True,
                cascade="all",
                backref=backref("parent", primaryjoin=node_table.c.parent_id==node_table.c.id, remote_side=node_table.c.id)
            ),
            'prev_sibling': relation(
                Node,
                primaryjoin=node_table.c.prev_sibling_id==node_table.c.id,
                remote_side=node_table.c.id,
                lazy=True,
                uselist=False
            ),
            'next_sibling': relation(
                Node,
                primaryjoin=node_table.c.next_sibling_id==node_table.c.id,
                remote_side=node_table.c.id,
                lazy=True,
                uselist=False,
                post_update=True
            )
        })

        session = create_session()

        def append_child(parent, child):
            if parent.children:
                parent.children[-1].next_sibling = child
                child.prev_sibling = parent.children[-1]
            parent.children.append(child)

        def remove_child(parent, child):
            child.parent = None
            node = child.next_sibling
            node.prev_sibling = child.prev_sibling
            child.prev_sibling.next_sibling = node
            session.delete(child)
        root = Node('root')

        about = Node('about')
        cats = Node('cats')
        stories = Node('stories')
        bruce = Node('bruce')

        append_child(root, about)
        assert(about.prev_sibling is None)
        append_child(root, cats)
        assert(cats.prev_sibling is about)
        assert(cats.next_sibling is None)
        assert(about.next_sibling is cats)
        assert(about.prev_sibling is None)
        append_child(root, stories)
        append_child(root, bruce)
        session.save(root)
        session.flush()

        remove_child(root, cats)
        # pre-trigger lazy loader on 'cats' to make the test easier
        cats.children
        self.assert_sql(testing.db, lambda: session.flush(), [
            (
                "UPDATE node SET prev_sibling_id=:prev_sibling_id WHERE node.id = :node_id",
                lambda ctx:{'prev_sibling_id':about.id, 'node_id':stories.id}
            ),
            (
                "UPDATE node SET next_sibling_id=:next_sibling_id WHERE node.id = :node_id",
                lambda ctx:{'next_sibling_id':stories.id, 'node_id':about.id}
            ),
            (
                "UPDATE node SET next_sibling_id=:next_sibling_id WHERE node.id = :node_id",
                lambda ctx:{'next_sibling_id':None, 'node_id':cats.id}
            ),
            (
                "DELETE FROM node WHERE node.id = :id",
                lambda ctx:[{'id':cats.id}]
            ),
        ])

class SelfReferentialPostUpdateTest2(TestBase, AssertsExecutionResults):
    def setUpAll(self):
        global metadata, a_table
        metadata = MetaData(testing.db)
        a_table = Table("a", metadata,
                Column("id", Integer(), primary_key=True),
                Column("fui", String(128)),
                Column("b", Integer(), ForeignKey("a.id")),
            )
        a_table.create()
    def tearDownAll(self):
        a_table.drop()
    def testbasic(self):
        """test that post_update remembers to be involved in update operations as well,
        since it replaces the normal dependency processing completely [ticket:413]"""
        class a(object):
            def __init__(self, fui):
                self.fui = fui

        mapper(a, a_table, properties={
            'foo': relation(a, remote_side=[a_table.c.id], post_update=True),
        })

        session = create_session()

        f1 = a("f1")
        session.save(f1)
        session.flush()

        f2 = a("f2")
        f2.foo = f1
        # at this point f1 is already inserted.  but we need post_update
        # to fire off anyway
        session.save(f2)
        session.flush()

        session.clear()
        f1 = session.query(a).get(f1.id)
        f2 = session.query(a).get(f2.id)
        assert f2.foo is f1

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