python类ISOLATION_LEVEL_SERIALIZABLE的实例源码-面圈网

test_connection.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 27 收藏 0 点赞 0 评论 0

def test_reset(self):
        conn = self.conn
        # switch session characteristics
        conn.autocommit = True
        conn.isolation_level = 'serializable'
        conn.readonly = True
        if self.conn.server_version >= 90100:
            conn.deferrable = False

        self.assertTrue(conn.autocommit)
        self.assertEqual(conn.isolation_level, ext.ISOLATION_LEVEL_SERIALIZABLE)
        self.assertTrue(conn.readonly is True)
        if self.conn.server_version >= 90100:
            self.assertTrue(conn.deferrable is False)

        conn.reset()
        # now the session characteristics should be reverted
        self.assertTrue(not conn.autocommit)
        self.assertEqual(conn.isolation_level, ext.ISOLATION_LEVEL_DEFAULT)
        self.assertTrue(conn.readonly is None)
        if self.conn.server_version >= 90100:
            self.assertTrue(conn.deferrable is None)

test_connection.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 24 收藏 0 点赞 0 评论 0

def test_set_isolation_level_default(self):
        conn = self.connect()
        curs = conn.cursor()

        conn.autocommit = True
        curs.execute("set default_transaction_isolation to 'read committed'")

        conn.autocommit = False
        conn.set_isolation_level(ext.ISOLATION_LEVEL_SERIALIZABLE)
        self.assertEqual(conn.isolation_level,
            ext.ISOLATION_LEVEL_SERIALIZABLE)
        curs.execute("show transaction_isolation")
        self.assertEqual(curs.fetchone()[0], "serializable")

        conn.rollback()
        conn.set_isolation_level(ext.ISOLATION_LEVEL_DEFAULT)
        curs.execute("show transaction_isolation")
        self.assertEqual(curs.fetchone()[0], "read committed")

test_connection.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 31 收藏 0 点赞 0 评论 0

def test_set_isolation_level_default(self):
        conn = self.connect()
        curs = conn.cursor()

        conn.autocommit = True
        curs.execute("set default_transaction_isolation to 'read committed'")

        conn.autocommit = False
        conn.set_isolation_level(ext.ISOLATION_LEVEL_SERIALIZABLE)
        self.assertEqual(conn.isolation_level,
            ext.ISOLATION_LEVEL_SERIALIZABLE)
        curs.execute("show transaction_isolation")
        self.assertEqual(curs.fetchone()[0], "serializable")

        conn.rollback()
        conn.set_isolation_level(ext.ISOLATION_LEVEL_DEFAULT)
        curs.execute("show transaction_isolation")
        self.assertEqual(curs.fetchone()[0], "read committed")

test_connection.py 文件源码项目：ShelbySearch 作者: Agentscreech 项目源码文件源码阅读 26 收藏 0 点赞 0 评论 0

def test_reset(self):
        conn = self.conn
        # switch session characteristics
        conn.autocommit = True
        conn.isolation_level = 'serializable'
        conn.readonly = True
        if self.conn.server_version >= 90100:
            conn.deferrable = False

        self.assertTrue(conn.autocommit)
        self.assertEqual(conn.isolation_level, ext.ISOLATION_LEVEL_SERIALIZABLE)
        self.assertTrue(conn.readonly is True)
        if self.conn.server_version >= 90100:
            self.assertTrue(conn.deferrable is False)

        conn.reset()
        # now the session characteristics should be reverted
        self.assertTrue(not conn.autocommit)
        self.assertEqual(conn.isolation_level, ext.ISOLATION_LEVEL_DEFAULT)
        self.assertTrue(conn.readonly is None)
        if self.conn.server_version >= 90100:
            self.assertTrue(conn.deferrable is None)

test_connection.py 文件源码项目：ShelbySearch 作者: Agentscreech 项目源码文件源码阅读 33 收藏 0 点赞 0 评论 0

def test_set_isolation_level_default(self):
        conn = self.connect()
        curs = conn.cursor()

        conn.autocommit = True
        curs.execute("set default_transaction_isolation to 'read committed'")

        conn.autocommit = False
        conn.set_isolation_level(ext.ISOLATION_LEVEL_SERIALIZABLE)
        self.assertEqual(conn.isolation_level,
            ext.ISOLATION_LEVEL_SERIALIZABLE)
        curs.execute("show transaction_isolation")
        self.assertEqual(curs.fetchone()[0], "serializable")

        conn.rollback()
        conn.set_isolation_level(ext.ISOLATION_LEVEL_DEFAULT)
        curs.execute("show transaction_isolation")
        self.assertEqual(curs.fetchone()[0], "read committed")

test_connection.py 文件源码项目：nmbs-realtime-feed 作者: datamindedbe 项目源码文件源码阅读 23 收藏 0 点赞 0 评论 0

def test_reset(self):
        conn = self.conn
        # switch session characteristics
        conn.autocommit = True
        conn.isolation_level = 'serializable'
        conn.readonly = True
        if self.conn.server_version >= 90100:
            conn.deferrable = False

        self.assert_(conn.autocommit)
        self.assertEqual(conn.isolation_level, ext.ISOLATION_LEVEL_SERIALIZABLE)
        self.assert_(conn.readonly is True)
        if self.conn.server_version >= 90100:
            self.assert_(conn.deferrable is False)

        conn.reset()
        # now the session characteristics should be reverted
        self.assert_(not conn.autocommit)
        self.assertEqual(conn.isolation_level, ext.ISOLATION_LEVEL_DEFAULT)
        self.assert_(conn.readonly is None)
        if self.conn.server_version >= 90100:
            self.assert_(conn.deferrable is None)

test_connection.py 文件源码项目：nmbs-realtime-feed 作者: datamindedbe 项目源码文件源码阅读 26 收藏 0 点赞 0 评论 0

def test_set_isolation_level_default(self):
        conn = self.connect()
        curs = conn.cursor()

        conn.autocommit = True
        curs.execute("set default_transaction_isolation to 'read committed'")

        conn.autocommit = False
        conn.set_isolation_level(ext.ISOLATION_LEVEL_SERIALIZABLE)
        self.assertEqual(conn.isolation_level,
            ext.ISOLATION_LEVEL_SERIALIZABLE)
        curs.execute("show transaction_isolation")
        self.assertEqual(curs.fetchone()[0], "serializable")

        conn.rollback()
        conn.set_isolation_level(ext.ISOLATION_LEVEL_DEFAULT)
        curs.execute("show transaction_isolation")
        self.assertEqual(curs.fetchone()[0], "read committed")

test_connection.py 文件源码项目：flask 作者: bobohope 项目源码文件源码阅读 25 收藏 0 点赞 0 评论 0

def test_reset(self):
        conn = self.conn
        # switch session characteristics
        conn.autocommit = True
        conn.isolation_level = 'serializable'
        conn.readonly = True
        if self.conn.server_version >= 90100:
            conn.deferrable = False

        self.assert_(conn.autocommit)
        self.assertEqual(conn.isolation_level, ext.ISOLATION_LEVEL_SERIALIZABLE)
        self.assert_(conn.readonly is True)
        if self.conn.server_version >= 90100:
            self.assert_(conn.deferrable is False)

        conn.reset()
        # now the session characteristics should be reverted
        self.assert_(not conn.autocommit)
        self.assertEqual(conn.isolation_level, ext.ISOLATION_LEVEL_DEFAULT)
        self.assert_(conn.readonly is None)
        if self.conn.server_version >= 90100:
            self.assert_(conn.deferrable is None)

test_connection.py 文件源码项目：flask 作者: bobohope 项目源码文件源码阅读 39 收藏 0 点赞 0 评论 0

def test_set_isolation_level_default(self):
        conn = self.connect()
        curs = conn.cursor()

        conn.autocommit = True
        curs.execute("set default_transaction_isolation to 'read committed'")

        conn.autocommit = False
        conn.set_isolation_level(ext.ISOLATION_LEVEL_SERIALIZABLE)
        self.assertEqual(conn.isolation_level,
            ext.ISOLATION_LEVEL_SERIALIZABLE)
        curs.execute("show transaction_isolation")
        self.assertEqual(curs.fetchone()[0], "serializable")

        conn.rollback()
        conn.set_isolation_level(ext.ISOLATION_LEVEL_DEFAULT)
        curs.execute("show transaction_isolation")
        self.assertEqual(curs.fetchone()[0], "read committed")

psycopg2.py 文件源码项目：Flask_Blog 作者: sugarguo 项目源码文件源码阅读 25 收藏 0 点赞 0 评论 0

def _isolation_lookup(self):
        extensions = self._psycopg2_extensions()
        return {
            'AUTOCOMMIT': extensions.ISOLATION_LEVEL_AUTOCOMMIT,
            'READ COMMITTED': extensions.ISOLATION_LEVEL_READ_COMMITTED,
            'READ UNCOMMITTED': extensions.ISOLATION_LEVEL_READ_UNCOMMITTED,
            'REPEATABLE READ': extensions.ISOLATION_LEVEL_REPEATABLE_READ,
            'SERIALIZABLE': extensions.ISOLATION_LEVEL_SERIALIZABLE
        }

test_connection.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 24 收藏 0 点赞 0 评论 0

def test_set_isolation_level(self):
        conn = self.connect()
        curs = conn.cursor()

        levels = [
            ('read uncommitted',
                ext.ISOLATION_LEVEL_READ_UNCOMMITTED),
            ('read committed', ext.ISOLATION_LEVEL_READ_COMMITTED),
            ('repeatable read', ext.ISOLATION_LEVEL_REPEATABLE_READ),
            ('serializable', ext.ISOLATION_LEVEL_SERIALIZABLE),
        ]
        for name, level in levels:
            conn.set_isolation_level(level)

            # the only values available on prehistoric PG versions
            if conn.server_version < 80000:
                if level in (
                        ext.ISOLATION_LEVEL_READ_UNCOMMITTED,
                        ext.ISOLATION_LEVEL_REPEATABLE_READ):
                    name, level = levels[levels.index((name, level)) + 1]

            self.assertEqual(conn.isolation_level, level)

            curs.execute('show transaction_isolation;')
            got_name = curs.fetchone()[0]

            self.assertEqual(name, got_name)
            conn.commit()

        self.assertRaises(ValueError, conn.set_isolation_level, -1)
        self.assertRaises(ValueError, conn.set_isolation_level, 5)

test_connection.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 27 收藏 0 点赞 0 评论 0

def test_set_isolation_level_autocommit(self):
        conn = self.connect()
        curs = conn.cursor()

        conn.set_isolation_level(ext.ISOLATION_LEVEL_AUTOCOMMIT)
        self.assertEqual(conn.isolation_level, ext.ISOLATION_LEVEL_DEFAULT)
        self.assertTrue(conn.autocommit)

        conn.isolation_level = 'serializable'
        self.assertEqual(conn.isolation_level, ext.ISOLATION_LEVEL_SERIALIZABLE)
        self.assertTrue(conn.autocommit)

        curs.execute('show transaction_isolation;')
        self.assertEqual(curs.fetchone()[0], 'serializable')

test_connection.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 36 收藏 0 点赞 0 评论 0

def test_set_isolation_level_abort(self):
        conn = self.connect()
        cur = conn.cursor()

        self.assertEqual(ext.TRANSACTION_STATUS_IDLE,
            conn.get_transaction_status())
        cur.execute("insert into isolevel values (10);")
        self.assertEqual(ext.TRANSACTION_STATUS_INTRANS,
            conn.get_transaction_status())

        conn.set_isolation_level(
            psycopg2.extensions.ISOLATION_LEVEL_SERIALIZABLE)
        self.assertEqual(psycopg2.extensions.TRANSACTION_STATUS_IDLE,
            conn.get_transaction_status())
        cur.execute("select count(*) from isolevel;")
        self.assertEqual(0, cur.fetchone()[0])

        cur.execute("insert into isolevel values (10);")
        self.assertEqual(psycopg2.extensions.TRANSACTION_STATUS_INTRANS,
            conn.get_transaction_status())
        conn.set_isolation_level(
            psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
        self.assertEqual(psycopg2.extensions.TRANSACTION_STATUS_IDLE,
            conn.get_transaction_status())
        cur.execute("select count(*) from isolevel;")
        self.assertEqual(0, cur.fetchone()[0])

        cur.execute("insert into isolevel values (10);")
        self.assertEqual(psycopg2.extensions.TRANSACTION_STATUS_IDLE,
            conn.get_transaction_status())
        conn.set_isolation_level(
            psycopg2.extensions.ISOLATION_LEVEL_READ_COMMITTED)
        self.assertEqual(psycopg2.extensions.TRANSACTION_STATUS_IDLE,
            conn.get_transaction_status())
        cur.execute("select count(*) from isolevel;")
        self.assertEqual(1, cur.fetchone()[0])
        self.assertEqual(conn.isolation_level,
            psycopg2.extensions.ISOLATION_LEVEL_READ_COMMITTED)

test_connection.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 27 收藏 0 点赞 0 评论 0

def test_closed(self):
        self.conn.close()
        self.assertRaises(psycopg2.InterfaceError,
            self.conn.set_session,
            ext.ISOLATION_LEVEL_SERIALIZABLE)

test_connection.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 27 收藏 0 点赞 0 评论 0

def test_not_in_transaction(self):
        cur = self.conn.cursor()
        cur.execute("select 1")
        self.assertRaises(psycopg2.ProgrammingError,
            self.conn.set_session,
            ext.ISOLATION_LEVEL_SERIALIZABLE)

test_connection.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 25 收藏 0 点赞 0 评论 0

def test_set_isolation_level(self):
        cur = self.conn.cursor()
        self.conn.set_session(
            ext.ISOLATION_LEVEL_SERIALIZABLE)
        cur.execute("SHOW transaction_isolation;")
        self.assertEqual(cur.fetchone()[0], 'serializable')
        self.conn.rollback()

        self.conn.set_session(
            ext.ISOLATION_LEVEL_REPEATABLE_READ)
        cur.execute("SHOW transaction_isolation;")
        if self.conn.server_version > 80000:
            self.assertEqual(cur.fetchone()[0], 'repeatable read')
        else:
            self.assertEqual(cur.fetchone()[0], 'serializable')
        self.conn.rollback()

        self.conn.set_session(
            isolation_level=ext.ISOLATION_LEVEL_READ_COMMITTED)
        cur.execute("SHOW transaction_isolation;")
        self.assertEqual(cur.fetchone()[0], 'read committed')
        self.conn.rollback()

        self.conn.set_session(
            isolation_level=ext.ISOLATION_LEVEL_READ_UNCOMMITTED)
        cur.execute("SHOW transaction_isolation;")
        if self.conn.server_version > 80000:
            self.assertEqual(cur.fetchone()[0], 'read uncommitted')
        else:
            self.assertEqual(cur.fetchone()[0], 'read committed')
        self.conn.rollback()

test_connection.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 24 收藏 0 点赞 0 评论 0

def test_set_isolation_level(self):
        conn = self.connect()
        curs = conn.cursor()

        levels = [
            ('read uncommitted',
                ext.ISOLATION_LEVEL_READ_UNCOMMITTED),
            ('read committed', ext.ISOLATION_LEVEL_READ_COMMITTED),
            ('repeatable read', ext.ISOLATION_LEVEL_REPEATABLE_READ),
            ('serializable', ext.ISOLATION_LEVEL_SERIALIZABLE),
        ]
        for name, level in levels:
            conn.set_isolation_level(level)

            # the only values available on prehistoric PG versions
            if conn.server_version < 80000:
                if level in (
                        ext.ISOLATION_LEVEL_READ_UNCOMMITTED,
                        ext.ISOLATION_LEVEL_REPEATABLE_READ):
                    name, level = levels[levels.index((name, level)) + 1]

            self.assertEqual(conn.isolation_level, level)

            curs.execute('show transaction_isolation;')
            got_name = curs.fetchone()[0]

            self.assertEqual(name, got_name)
            conn.commit()

        self.assertRaises(ValueError, conn.set_isolation_level, -1)
        self.assertRaises(ValueError, conn.set_isolation_level, 5)

test_connection.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 24 收藏 0 点赞 0 评论 0

def test_set_isolation_level_autocommit(self):
        conn = self.connect()
        curs = conn.cursor()

        conn.set_isolation_level(ext.ISOLATION_LEVEL_AUTOCOMMIT)
        self.assertEqual(conn.isolation_level, ext.ISOLATION_LEVEL_DEFAULT)
        self.assert_(conn.autocommit)

        conn.isolation_level = 'serializable'
        self.assertEqual(conn.isolation_level, ext.ISOLATION_LEVEL_SERIALIZABLE)
        self.assert_(conn.autocommit)

        curs.execute('show transaction_isolation;')
        self.assertEqual(curs.fetchone()[0], 'serializable')

test_connection.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 44 收藏 0 点赞 0 评论 0

def test_set_isolation_level_abort(self):
        conn = self.connect()
        cur = conn.cursor()

        self.assertEqual(ext.TRANSACTION_STATUS_IDLE,
            conn.get_transaction_status())
        cur.execute("insert into isolevel values (10);")
        self.assertEqual(ext.TRANSACTION_STATUS_INTRANS,
            conn.get_transaction_status())

        conn.set_isolation_level(
            psycopg2.extensions.ISOLATION_LEVEL_SERIALIZABLE)
        self.assertEqual(psycopg2.extensions.TRANSACTION_STATUS_IDLE,
            conn.get_transaction_status())
        cur.execute("select count(*) from isolevel;")
        self.assertEqual(0, cur.fetchone()[0])

        cur.execute("insert into isolevel values (10);")
        self.assertEqual(psycopg2.extensions.TRANSACTION_STATUS_INTRANS,
            conn.get_transaction_status())
        conn.set_isolation_level(
            psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
        self.assertEqual(psycopg2.extensions.TRANSACTION_STATUS_IDLE,
            conn.get_transaction_status())
        cur.execute("select count(*) from isolevel;")
        self.assertEqual(0, cur.fetchone()[0])

        cur.execute("insert into isolevel values (10);")
        self.assertEqual(psycopg2.extensions.TRANSACTION_STATUS_IDLE,
            conn.get_transaction_status())
        conn.set_isolation_level(
            psycopg2.extensions.ISOLATION_LEVEL_READ_COMMITTED)
        self.assertEqual(psycopg2.extensions.TRANSACTION_STATUS_IDLE,
            conn.get_transaction_status())
        cur.execute("select count(*) from isolevel;")
        self.assertEqual(1, cur.fetchone()[0])
        self.assertEqual(conn.isolation_level,
            psycopg2.extensions.ISOLATION_LEVEL_READ_COMMITTED)

test_connection.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 59 收藏 0 点赞 0 评论 0

def test_isolation_level_serializable(self):
        cnn1 = self.connect()
        cnn2 = self.connect()
        cnn2.set_isolation_level(ext.ISOLATION_LEVEL_SERIALIZABLE)

        cur1 = cnn1.cursor()
        cur1.execute("select count(*) from isolevel;")
        self.assertEqual(0, cur1.fetchone()[0])
        cnn1.commit()

        cur2 = cnn2.cursor()
        cur2.execute("insert into isolevel values (10);")
        cur1.execute("insert into isolevel values (20);")

        cur2.execute("select count(*) from isolevel;")
        self.assertEqual(1, cur2.fetchone()[0])
        cnn1.commit()
        cur2.execute("select count(*) from isolevel;")
        self.assertEqual(1, cur2.fetchone()[0])

        cur1.execute("select count(*) from isolevel;")
        self.assertEqual(1, cur1.fetchone()[0])
        cnn2.commit()
        cur1.execute("select count(*) from isolevel;")
        self.assertEqual(2, cur1.fetchone()[0])

        cur2.execute("select count(*) from isolevel;")
        self.assertEqual(2, cur2.fetchone()[0])

test_connection.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 31 收藏 0 点赞 0 评论 0

def test_closed(self):
        self.conn.close()
        self.assertRaises(psycopg2.InterfaceError,
            self.conn.set_session,
            ext.ISOLATION_LEVEL_SERIALIZABLE)

test_connection.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 27 收藏 0 点赞 0 评论 0

def test_not_in_transaction(self):
        cur = self.conn.cursor()
        cur.execute("select 1")
        self.assertRaises(psycopg2.ProgrammingError,
            self.conn.set_session,
            ext.ISOLATION_LEVEL_SERIALIZABLE)

test_connection.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 27 收藏 0 点赞 0 评论 0

def test_set_isolation_level(self):
        cur = self.conn.cursor()
        self.conn.set_session(
            ext.ISOLATION_LEVEL_SERIALIZABLE)
        cur.execute("SHOW transaction_isolation;")
        self.assertEqual(cur.fetchone()[0], 'serializable')
        self.conn.rollback()

        self.conn.set_session(
            ext.ISOLATION_LEVEL_REPEATABLE_READ)
        cur.execute("SHOW transaction_isolation;")
        if self.conn.server_version > 80000:
            self.assertEqual(cur.fetchone()[0], 'repeatable read')
        else:
            self.assertEqual(cur.fetchone()[0], 'serializable')
        self.conn.rollback()

        self.conn.set_session(
            isolation_level=ext.ISOLATION_LEVEL_READ_COMMITTED)
        cur.execute("SHOW transaction_isolation;")
        self.assertEqual(cur.fetchone()[0], 'read committed')
        self.conn.rollback()

        self.conn.set_session(
            isolation_level=ext.ISOLATION_LEVEL_READ_UNCOMMITTED)
        cur.execute("SHOW transaction_isolation;")
        if self.conn.server_version > 80000:
            self.assertEqual(cur.fetchone()[0], 'read uncommitted')
        else:
            self.assertEqual(cur.fetchone()[0], 'read committed')
        self.conn.rollback()

psycopg2.py 文件源码项目：QXSConsolas 作者: qxsch 项目源码文件源码阅读 23 收藏 0 点赞 0 评论 0

def _isolation_lookup(self):
        extensions = self._psycopg2_extensions()
        return {
            'AUTOCOMMIT': extensions.ISOLATION_LEVEL_AUTOCOMMIT,
            'READ COMMITTED': extensions.ISOLATION_LEVEL_READ_COMMITTED,
            'READ UNCOMMITTED': extensions.ISOLATION_LEVEL_READ_UNCOMMITTED,
            'REPEATABLE READ': extensions.ISOLATION_LEVEL_REPEATABLE_READ,
            'SERIALIZABLE': extensions.ISOLATION_LEVEL_SERIALIZABLE
        }

psycopg2.py 文件源码项目：flasky 作者: RoseOu 项目源码文件源码阅读 25 收藏 0 点赞 0 评论 0

def _isolation_lookup(self):
        from psycopg2 import extensions
        return {
            'AUTOCOMMIT': extensions.ISOLATION_LEVEL_AUTOCOMMIT,
            'READ COMMITTED': extensions.ISOLATION_LEVEL_READ_COMMITTED,
            'READ UNCOMMITTED': extensions.ISOLATION_LEVEL_READ_UNCOMMITTED,
            'REPEATABLE READ': extensions.ISOLATION_LEVEL_REPEATABLE_READ,
            'SERIALIZABLE': extensions.ISOLATION_LEVEL_SERIALIZABLE
        }

psycopg2.py 文件源码项目：oa_qian 作者: sunqb 项目源码文件源码阅读 23 收藏 0 点赞 0 评论 0

def _isolation_lookup(self):
        from psycopg2 import extensions
        return {
            'AUTOCOMMIT': extensions.ISOLATION_LEVEL_AUTOCOMMIT,
            'READ COMMITTED': extensions.ISOLATION_LEVEL_READ_COMMITTED,
            'READ UNCOMMITTED': extensions.ISOLATION_LEVEL_READ_UNCOMMITTED,
            'REPEATABLE READ': extensions.ISOLATION_LEVEL_REPEATABLE_READ,
            'SERIALIZABLE': extensions.ISOLATION_LEVEL_SERIALIZABLE
        }

psycopg2.py 文件源码项目：chihu 作者: yelongyu 项目源码文件源码阅读 22 收藏 0 点赞 0 评论 0

def _isolation_lookup(self):
        extensions = self._psycopg2_extensions()
        return {
            'AUTOCOMMIT': extensions.ISOLATION_LEVEL_AUTOCOMMIT,
            'READ COMMITTED': extensions.ISOLATION_LEVEL_READ_COMMITTED,
            'READ UNCOMMITTED': extensions.ISOLATION_LEVEL_READ_UNCOMMITTED,
            'REPEATABLE READ': extensions.ISOLATION_LEVEL_REPEATABLE_READ,
            'SERIALIZABLE': extensions.ISOLATION_LEVEL_SERIALIZABLE
        }

psycopg2.py 文件源码项目：ShelbySearch 作者: Agentscreech 项目源码文件源码阅读 22 收藏 0 点赞 0 评论 0

def _isolation_lookup(self):
        extensions = self._psycopg2_extensions()
        return {
            'AUTOCOMMIT': extensions.ISOLATION_LEVEL_AUTOCOMMIT,
            'READ COMMITTED': extensions.ISOLATION_LEVEL_READ_COMMITTED,
            'READ UNCOMMITTED': extensions.ISOLATION_LEVEL_READ_UNCOMMITTED,
            'REPEATABLE READ': extensions.ISOLATION_LEVEL_REPEATABLE_READ,
            'SERIALIZABLE': extensions.ISOLATION_LEVEL_SERIALIZABLE
        }

test_connection.py 文件源码项目：ShelbySearch 作者: Agentscreech 项目源码文件源码阅读 27 收藏 0 点赞 0 评论 0

def test_set_isolation_level(self):
        conn = self.connect()
        curs = conn.cursor()

        levels = [
            ('read uncommitted',
                ext.ISOLATION_LEVEL_READ_UNCOMMITTED),
            ('read committed', ext.ISOLATION_LEVEL_READ_COMMITTED),
            ('repeatable read', ext.ISOLATION_LEVEL_REPEATABLE_READ),
            ('serializable', ext.ISOLATION_LEVEL_SERIALIZABLE),
        ]
        for name, level in levels:
            conn.set_isolation_level(level)

            # the only values available on prehistoric PG versions
            if conn.server_version < 80000:
                if level in (
                        ext.ISOLATION_LEVEL_READ_UNCOMMITTED,
                        ext.ISOLATION_LEVEL_REPEATABLE_READ):
                    name, level = levels[levels.index((name, level)) + 1]

            self.assertEqual(conn.isolation_level, level)

            curs.execute('show transaction_isolation;')
            got_name = curs.fetchone()[0]

            self.assertEqual(name, got_name)
            conn.commit()

        self.assertRaises(ValueError, conn.set_isolation_level, -1)
        self.assertRaises(ValueError, conn.set_isolation_level, 5)

test_connection.py 文件源码项目：ShelbySearch 作者: Agentscreech 项目源码文件源码阅读 27 收藏 0 点赞 0 评论 0

def test_set_isolation_level_autocommit(self):
        conn = self.connect()
        curs = conn.cursor()

        conn.set_isolation_level(ext.ISOLATION_LEVEL_AUTOCOMMIT)
        self.assertEqual(conn.isolation_level, ext.ISOLATION_LEVEL_DEFAULT)
        self.assertTrue(conn.autocommit)

        conn.isolation_level = 'serializable'
        self.assertEqual(conn.isolation_level, ext.ISOLATION_LEVEL_SERIALIZABLE)
        self.assertTrue(conn.autocommit)

        curs.execute('show transaction_isolation;')
        self.assertEqual(curs.fetchone()[0], 'serializable')