session.py

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

        sess.begin_nested()
        
        u1 = User()
        sess.save(u1)
        sess.flush()
        
        sess.rollback()
        
        u2 = User()
        sess.save(u2)
        
        sess.commit()
        
        self.assertEquals(util.Set(sess.query(User).all()), util.Set([u2]))
        
        sess.begin()
        sess.begin_nested()
        
        u3 = User()
        sess.save(u3)
        sess.commit() # commit the nested transaction
        sess.rollback()
        
        self.assertEquals(util.Set(sess.query(User).all()), util.Set([u2]))
        
        sess.close()
    
    @testing.unsupported('sqlite', 'mssql', 'firebird', 'sybase', 'access',
                         'oracle', 'maxdb')
    @testing.exclude('mysql', '<', (5, 0, 3))
    def test_mixed_transaction_control(self):
        class User(object): pass
        mapper(User, users)
        
        sess = create_session(transactional=False)
        
        sess.begin()
        sess.begin_nested()
        transaction = sess.begin()
    
        sess.save(User())
        
        transaction.commit()
        sess.commit()
        sess.commit()
        
        sess.close()
        
        self.assertEquals(len(sess.query(User).all()), 1)
        
        t1 = sess.begin()
        t2 = sess.begin_nested()
    
        sess.save(User())
        
        t2.commit()
        assert sess.transaction is t1
        
        sess.close()
    
    @testing.unsupported('sqlite', 'mssql', 'firebird', 'sybase', 'access',
                         'oracle', 'maxdb')
    @testing.exclude('mysql', '<', (5, 0, 3))
    def test_mixed_transaction_close(self):
        class User(object): pass
        mapper(User, users)
        
        sess = create_session(transactional=True)
        
        sess.begin_nested()
    
        sess.save(User())
        sess.flush()
        
        sess.close()
        
        sess.save(User())
        sess.commit()
        
        sess.close()
        
        self.assertEquals(len(sess.query(User).all()), 1)
    
    @testing.unsupported('sqlite', 'mssql', 'firebird', 'sybase', 'access',
                         'oracle', 'maxdb')
    @testing.exclude('mysql', '<', (5, 0, 3))
    def test_error_on_using_inactive_session(self):
        class User(object): pass
        mapper(User, users)
        
        sess = create_session(transactional=False)
        
        try:
            sess.begin()
            sess.begin()
        
            sess.save(User())
            sess.flush()
            
            sess.rollback()
            sess.begin()
            assert False
        except exceptions.InvalidRequestError, e:
            self.assertEquals(str(e), "The transaction is inactive due to a rollback in a subtransaction and should be closed")
        sess.close()
    
    @engines.close_open_connections
    def test_bound_connection(self):
        class User(object):pass
        mapper(User, users)
        c = testing.db.connect()
        sess = create_session(bind=c)
        sess.create_transaction()
        transaction = sess.transaction
        u = User()
        sess.save(u)
        sess.flush()
        assert transaction.get_or_add(testing.db) is transaction.get_or_add(c) is c

        try:
            transaction.add(testing.db.connect())
            assert False
        except exceptions.InvalidRequestError, e:
            assert str(e) == "Session already has a Connection associated for the given Connection's Engine"

        try:
            transaction.get_or_add(testing.db.connect())
            assert False
        except exceptions.InvalidRequestError, e:
            assert str(e) == "Session already has a Connection associated for the given Connection's Engine"

        try:
            transaction.add(testing.db)
            assert False
        except exceptions.InvalidRequestError, e:
            assert str(e) == "Session already has a Connection associated for the given Engine"

        transaction.rollback()
        assert len(sess.query(User).all()) == 0
        sess.close()

    def test_bound_connection_transactional(self):
        class User(object):pass
        mapper(User, users)
        c = testing.db.connect()

        sess = create_session(bind=c, transactional=True)
        u = User()
        sess.save(u)
        sess.flush()
        sess.close()
        assert not c.in_transaction()
        assert c.scalar("select count(1) from users") == 0

        sess = create_session(bind=c, transactional=True)
        u = User()
        sess.save(u)
        sess.flush()
        sess.commit()
        assert not c.in_transaction()
        assert c.scalar("select count(1) from users") == 1
        c.execute("delete from users")
        assert c.scalar("select count(1) from users") == 0

        c = testing.db.connect()

        trans = c.begin()
        sess = create_session(bind=c, transactional=False)
        u = User()
        sess.save(u)
        sess.flush()
        assert c.in_transaction()
        trans.commit()
        assert not c.in_transaction()
        assert c.scalar("select count(1) from users") == 1


    @engines.close_open_connections
    def test_save_update_delete(self):

        s = create_session()
        class User(object):
            pass
        mapper(User, users)

        user = User()

        try:
            s.update(user)
            assert False
        except exceptions.InvalidRequestError, e:
            assert str(e) == "Instance 'User@%s' is not persisted" % hex(id(user))

        try:
            s.delete(user)
            assert False
        except exceptions.InvalidRequestError, e:
            assert str(e) == "Instance 'User@%s' is not persisted" % hex(id(user))

        s.save(user)
        s.flush()
        user = s.query(User).one()
        s.expunge(user)
        assert user not in s

        # modify outside of session, assert changes remain/get saved
        user.user_name = "fred"
        s.update(user)
        assert user in s
        assert user in s.dirty
        s.flush()
        s.clear()
        assert s.query(User).count() == 1
        user = s.query(User).one()
        assert user.user_name == 'fred'

        # ensure its not dirty if no changes occur
        s.clear()
        assert user not in s
        s.update(user)
        assert user in s
        assert user not in s.dirty

        try:
            s.save(user)
            assert False
        except exceptions.InvalidRequestError, e:
            assert str(e) == "Instance 'User@%s' is already persistent" % hex(id(user))

        s2 = create_session()
        try:
            s2.delete(user)
            assert False
        except exceptions.InvalidRequestError, e:
            assert "is already attached to session" in str(e)

        u2 = s2.query(User).get(user.user_id)
        try:
            s.delete(u2)
            assert False
        except exceptions.InvalidRequestError, e:
            assert "already persisted with a different identity" in str(e)

        s.delete(user)
        s.flush()
        assert user not in s
        assert s.query(User).count() == 0

    def test_is_modified(self):
        s = create_session()
        class User(object):pass
        class Address(object):pass

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

        # save user
        u = User()
        u.user_name = 'fred'
        s.save(u)
        s.flush()
        s.clear()

        user = s.query(User).one()
        assert user not in s.dirty
        assert not s.is_modified(user)
        user.user_name = 'fred'
        assert user in s.dirty
        assert not s.is_modified(user)
        user.user_name = 'ed'
        assert user in s.dirty
        assert s.is_modified(user)
        s.flush()
        assert user not in s.dirty
        assert not s.is_modified(user)

        a = Address()
        user.addresses.append(a)
        assert user in s.dirty
        assert s.is_modified(user)
        assert not s.is_modified(user, include_collections=False)


    def test_weak_ref(self):
        """test the weak-referencing identity map, which strongly-references modified items."""

        s = create_session()
        class User(fixtures.Base):pass
        mapper(User, users)

        s.save(User(user_name='ed'))
        s.flush()
        assert not s.dirty

        user = s.query(User).one()
        del user
        gc.collect()
        assert len(s.identity_map) == 0
        assert len(s.identity_map.data) == 0

        user = s.query(User).one()
        user.user_name = 'fred'
        del user
        gc.collect()
        assert len(s.identity_map) == 1
        assert len(s.identity_map.data) == 1
        assert len(s.dirty) == 1

        s.flush()
        gc.collect()
        assert not s.dirty
        assert not s.identity_map
        assert not s.identity_map.data

        user = s.query(User).one()
        assert user.user_name == 'fred'
        assert s.identity_map

    def test_strong_ref(self):
        s = create_session(weak_identity_map=False)
        class User(object):pass
        mapper(User, users)

        # save user
        s.save(User())
        s.flush()
        user = s.query(User).one()
        user = None
        print s.identity_map
        import gc
        gc.collect()
        assert len(s.identity_map) == 1

    def test_prune(self):
        s = create_session(weak_identity_map=False)
        class User(object):pass
        mapper(User, users)

        for o in [User() for x in xrange(10)]:
            s.save(o)
        # o is still live after this loop...

        self.assert_(len(s.identity_map) == 0)
        self.assert_(s.prune() == 0)
        s.flush()
        import gc
        gc.collect()
        self.assert_(s.prune() == 9)
        self.assert_(len(s.identity_map) == 1)

        user_id = o.user_id
        del o
        self.assert_(s.prune() == 1)
        self.assert_(len(s.identity_map) == 0)

        u = s.query(User).get(user_id)
        self.assert_(s.prune() == 0)
        self.assert_(len(s.identity_map) == 1)
        u.user_name = 'squiznart'
        del u
        self.assert_(s.prune() == 0)
        self.assert_(len(s.identity_map) == 1)
        s.flush()
        self.assert_(s.prune() == 1)
        self.assert_(len(s.identity_map) == 0)

        s.save(User())