unitofwork.py

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

        """tests that when a lazy-loaded list is unloaded, and a commit occurs, that the
        'passive' call on that list does not blow away its value"""

        m1 = mapper(User, users, properties = {
            'addresses': relation(mapper(Address, addresses))
        })

        u = User()
        u.addresses.append(Address())
        u.addresses.append(Address())
        u.addresses.append(Address())
        u.addresses.append(Address())
        Session.commit()
        Session.close()
        ulist = Session.query(m1).all()
        u1 = ulist[0]
        u1.user_name = 'newname'
        Session.commit()
        self.assert_(len(u1.addresses) == 4)

    def test_inherits(self):
        m1 = mapper(User, users)

        class AddressUser(User):
            """a user object that also has the users mailing address."""
            pass

        # define a mapper for AddressUser that inherits the User.mapper, and joins on the user_id column
        AddressUser.mapper = mapper(
                AddressUser,
                addresses, inherits=m1
                )

        au = AddressUser()
        Session.commit()
        Session.close()
        l = Session.query(AddressUser).one()
        self.assert_(l.user_id == au.user_id and l.address_id == au.address_id)

    def test_deferred(self):
        """test deferred column operations"""

        mapper(User, users, properties={
            'user_name':deferred(users.c.user_name)
        })

        # dont set deferred attribute, commit session
        u = User()
        u.user_id=42
        Session.commit()

        #  assert that changes get picked up
        u.user_name = 'some name'
        Session.commit()
        assert list(Session.execute(users.select(), mapper=User)) == [(42, 'some name')]
        Session.clear()

        # assert that a set operation doesn't trigger a load operation
        u = Session.query(User).filter(User.user_name=='some name').one()
        def go():
            u.user_name = 'some other name'
        self.assert_sql_count(testing.db, go, 0)
        Session.flush()
        assert list(Session.execute(users.select(), mapper=User)) == [(42, 'some other name')]

        Session.clear()

        # test assigning None to an unloaded deferred also works
        u = Session.query(User).filter(User.user_name=='some other name').one()
        u.user_name = None
        Session.flush()
        assert list(Session.execute(users.select(), mapper=User)) == [(42, None)]


    # why no support on oracle ?  because oracle doesn't save
    # "blank" strings; it saves a single space character.
    @testing.unsupported('oracle')
    def test_dont_update_blanks(self):
        mapper(User, users)
        u = User()
        u.user_name = ""
        Session.commit()
        Session.close()
        u = Session.query(User).get(u.user_id)
        u.user_name = ""
        def go():
            Session.commit()
        self.assert_sql_count(testing.db, go, 0)

    def test_multitable(self):
        """tests a save of an object where each instance spans two tables. also tests
        redefinition of the keynames for the column properties."""
        usersaddresses = sql.join(users, addresses, users.c.user_id == addresses.c.user_id)
        m = mapper(User, usersaddresses,
            properties = dict(
                email = addresses.c.email_address,
                foo_id = [users.c.user_id, addresses.c.user_id],
                )
            )

        u = User()
        u.user_name = 'multitester'
        u.email = '[e-mail]'

        Session.commit()
        id = m.primary_key_from_instance(u)

        Session.close()

        u = Session.get(User, id)
        assert u.user_name == 'multitester'

        usertable = users.select(users.c.user_id.in_([u.foo_id])).execute().fetchall()
        self.assertEqual(usertable[0].values(), [u.foo_id, 'multitester'])
        addresstable = addresses.select(addresses.c.address_id.in_([u.address_id])).execute().fetchall()
        self.assertEqual(addresstable[0].values(), [u.address_id, u.foo_id, '[e-mail]'])

        u.email = '[e-mail]'
        u.user_name = 'imnew'
        Session.commit()

        usertable = users.select(users.c.user_id.in_([u.foo_id])).execute().fetchall()
        self.assertEqual(usertable[0].values(), [u.foo_id, 'imnew'])
        addresstable = addresses.select(addresses.c.address_id.in_([u.address_id])).execute().fetchall()
        self.assertEqual(addresstable[0].values(), [u.address_id, u.foo_id, '[e-mail]'])

        Session.close()
        u = Session.get(User, id)
        assert u.user_name == 'imnew'

    def test_history_get(self):
        """tests that the history properly lazy-fetches data when it wasnt otherwise loaded"""
        mapper(User, users, properties={
            'addresses':relation(Address, cascade="all, delete-orphan")
        })
        mapper(Address, addresses)

        u = User()
        u.addresses.append(Address())
        u.addresses.append(Address())
        Session.commit()
        Session.close()
        u = Session.query(User).get(u.user_id)
        Session.delete(u)
        Session.commit()
        assert users.count().scalar() == 0
        assert addresses.count().scalar() == 0



    def test_batchmode(self):
        """test the 'batch=False' flag on mapper()"""

        class TestExtension(MapperExtension):
            def before_insert(self, mapper, connection, instance):
                self.current_instance = instance
            def after_insert(self, mapper, connection, instance):
                assert instance is self.current_instance
        m = mapper(User, users, extension=TestExtension(), batch=False)
        u1 = User()
        u1.username = 'user1'
        u2 = User()
        u2.username = 'user2'
        Session.commit()

        clear_mappers()

        m = mapper(User, users, extension=TestExtension())
        u1 = User()
        u1.username = 'user1'
        u2 = User()
        u2.username = 'user2'
        try:
            Session.commit()
            assert False
        except AssertionError:
            assert True


class ManyToOneTest(ORMTest):
    metadata = tables.metadata

    def define_tables(self, metadata):
        pass

    def test_m2o_onetoone(self):
        # TODO: put assertion in here !!!
        m = mapper(Address, addresses, properties = dict(
            user = relation(mapper(User, users), lazy = True, uselist = False)
        ))
        data = [
            {'user_name' : 'thesub' , 'email_address' : '[e-mail]'},
            {'user_name' : 'assdkfj' , 'email_address' : '[e-mail]'},
            {'user_name' : 'n4knd' , 'email_address' : '[e-mail]'},
            {'user_name' : 'v88f4' , 'email_address' : '[e-mail]'},
            {'user_name' : 'asdf8d' , 'email_address' : '[e-mail]'}
        ]
        objects = []
        for elem in data:
            a = Address()
            a.email_address = elem['email_address']
            a.user = User()
            a.user.user_name = elem['user_name']
            objects.append(a)

        Session.commit()
        objects[2].email_address = '[e-mail]'
        objects[3].user = User()
        objects[3].user.user_name = 'imnewlyadded'
        self.assert_sql(testing.db, lambda: Session.commit(), [
                (
                    "INSERT INTO users (user_name) VALUES (:user_name)",
                    {'user_name': 'imnewlyadded'}
                ),
                {
                    "UPDATE email_addresses SET email_address=:email_address WHERE email_addresses.address_id = :email_addresses_address_id":
                    lambda ctx: {'email_address': '[e-mail]', 'email_addresses_address_id': objects[2].address_id}
                ,

                    "UPDATE email_addresses SET user_id=:user_id WHERE email_addresses.address_id = :email_addresses_address_id":
                    lambda ctx: {'user_id': objects[3].user.user_id, 'email_addresses_address_id': objects[3].address_id}
                },

        ],
        with_sequences=[
                (
                    "INSERT INTO users (user_id, user_name) VALUES (:user_id, :user_name)",
                    lambda ctx:{'user_name': 'imnewlyadded', 'user_id':ctx.last_inserted_ids()[0]}
                ),
                {
                    "UPDATE email_addresses SET email_address=:email_address WHERE email_addresses.address_id = :email_addresses_address_id":
                    lambda ctx: {'email_address': '[e-mail]', 'email_addresses_address_id': objects[2].address_id}
                ,

                    "UPDATE email_addresses SET user_id=:user_id WHERE email_addresses.address_id = :email_addresses_address_id":
                    lambda ctx: {'user_id': objects[3].user.user_id, 'email_addresses_address_id': objects[3].address_id}
                },

        ])
        l = sql.select([users, addresses], sql.and_(users.c.user_id==addresses.c.user_id, addresses.c.address_id==a.address_id)).execute()
        assert l.fetchone().values() == [a.user.user_id, 'asdf8d', a.address_id, a.user_id, '[e-mail]']


    def test_manytoone_1(self):
        m = mapper(Address, addresses, properties = dict(
            user = relation(mapper(User, users), lazy = True)
        ))
        a1 = Address()
        a1.email_address = 'emailaddress1'
        u1 = User()
        u1.user_name='user1'

        a1.user = u1
        Session.commit()
        Session.close()
        a1 = Session.query(Address).get(a1.address_id)
        u1 = Session.query(User).get(u1.user_id)
        assert a1.user is u1

        a1.user = None
        Session.commit()
        Session.close()
        a1 = Session.query(Address).get(a1.address_id)
        u1 = Session.query(User).get(u1.user_id)
        assert a1.user is None

    def test_manytoone_2(self):
        m = mapper(Address, addresses, properties = dict(
            user = relation(mapper(User, users), lazy = True)
        ))
        a1 = Address()
        a1.email_address = 'emailaddress1'
        a2 = Address()
        a2.email_address = 'emailaddress2'
        u1 = User()
        u1.user_name='user1'

        a1.user = u1
        Session.commit()
        Session.close()
        a1 = Session.query(Address).get(a1.address_id)
        a2 = Session.query(Address).get(a2.address_id)
        u1 = Session.query(User).get(u1.user_id)
        assert a1.user is u1
        a1.user = None
        a2.user = u1
        Session.commit()
        Session.close()
        a1 = Session.query(Address).get(a1.address_id)
        a2 = Session.query(Address).get(a2.address_id)
        u1 = Session.query(User).get(u1.user_id)
        assert a1.user is None
        assert a2.user is u1

    def test_manytoone_3(self):
        m = mapper(Address, addresses, properties = dict(
            user = relation(mapper(User, users), lazy = True)
        ))
        a1 = Address()
        a1.email_address = 'emailaddress1'
        u1 = User()
        u1.user_name='user1'
        u2 = User()
        u2.user_name='user2'

        a1.user = u1
        Session.commit()
        Session.close()
        a1 = Session.query(Address).get(a1.address_id)
        u1 = Session.query(User).get(u1.user_id)
        u2 = Session.query(User).get(u2.user_id)
        assert a1.user is u1

        a1.user = u2
        Session.commit()
        Session.close()
        a1 = Session.query(Address).get(a1.address_id)
        u1 = Session.query(User).get(u1.user_id)
        u2 = Session.query(User).get(u2.user_id)
        assert a1.user is u2

    def test_bidirectional_noload(self):
        mapper(User, users, properties={
            'addresses':relation(Address, backref='user', lazy=None)
        })
        mapper(Address, addresses)

        sess = Session()

        # try it on unsaved objects
        u1 = User()
        a1 = Address()
        a1.user = u1
        sess.save(u1)
        sess.flush()
        sess.clear()

        a1 = sess.query(Address).get(a1.address_id)

        a1.user = None
        sess.flush()
        sess.clear()
        assert sess.query(Address).get(a1.address_id).user is None
        assert sess.query(User).get(u1.user_id).addresses == []


class ManyToManyTest(ORMTest):
    metadata = tables.metadata

    def define_tables(self, metadata):
        pass

    def test_manytomany(self):
        items = orderitems

        keywordmapper = mapper(Keyword, keywords)

        m = mapper(Item, items, properties = dict(
                keywords = relation(keywordmapper, itemkeywords, lazy = False, order_by=keywords.c.name),
            ))

        data = [Item,
            {'item_name': 'mm_item1', 'keywords' : (Keyword,[{'name': 'big'},{'name': 'green'}, {'name': 'purple'},{'name': 'round'}])},
            {'item_name': 'mm_item2', 'keywords' : (Keyword,[{'name':'blue'}, {'name':'imnew'},{'name':'round'}, {'name':'small'}])},
            {'item_name': 'mm_item3', 'keywords' : (Keyword,[])},
            {'item_name': 'mm_item4', 'keywords' : (Keyword,[{'name':'big'}, {'name':'blue'},])},
            {'item_name': 'mm_item5', 'keywords' : (Keyword,[{'name':'big'},{'name':'exacting'},{'name':'green'}])},
            {'item_name': 'mm_item6', 'keywords' : (Keyword,[{'name':'red'},{'name':'round'},{'name':'small'}])},