我们从Python开源项目中，提取了以下46个代码示例，用于说明如何使用psycopg2.extensions.ISOLATION_LEVEL_READ_UNCOMMITTED。
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 }
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)
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()
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 }
def _isolation_lookup(self): extensions = __import__('psycopg2.extensions').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 }
def commit(self): # print self.get_transaction_status_string() self._psql_conn.commit() # self._psql_conn.set_isolation_level(ISOLATION_LEVEL_READ_UNCOMMITTED)
def test_setattr_isolation_level_int(self): cur = self.conn.cursor() self.conn.isolation_level = ext.ISOLATION_LEVEL_SERIALIZABLE self.assertEqual(self.conn.isolation_level, ext.ISOLATION_LEVEL_SERIALIZABLE) cur.execute("SHOW transaction_isolation;") self.assertEqual(cur.fetchone()[0], 'serializable') self.conn.rollback() self.conn.isolation_level = ext.ISOLATION_LEVEL_REPEATABLE_READ cur.execute("SHOW transaction_isolation;") if self.conn.server_version > 80000: self.assertEqual(self.conn.isolation_level, ext.ISOLATION_LEVEL_REPEATABLE_READ) self.assertEqual(cur.fetchone()[0], 'repeatable read') else: self.assertEqual(self.conn.isolation_level, ext.ISOLATION_LEVEL_SERIALIZABLE) self.assertEqual(cur.fetchone()[0], 'serializable') self.conn.rollback() self.conn.isolation_level = ext.ISOLATION_LEVEL_READ_COMMITTED self.assertEqual(self.conn.isolation_level, ext.ISOLATION_LEVEL_READ_COMMITTED) cur.execute("SHOW transaction_isolation;") self.assertEqual(cur.fetchone()[0], 'read committed') self.conn.rollback() self.conn.isolation_level = ext.ISOLATION_LEVEL_READ_UNCOMMITTED cur.execute("SHOW transaction_isolation;") if self.conn.server_version > 80000: self.assertEqual(self.conn.isolation_level, ext.ISOLATION_LEVEL_READ_UNCOMMITTED) self.assertEqual(cur.fetchone()[0], 'read uncommitted') else: self.assertEqual(self.conn.isolation_level, ext.ISOLATION_LEVEL_READ_COMMITTED) self.assertEqual(cur.fetchone()[0], 'read committed') self.conn.rollback() self.assertEqual(ext.ISOLATION_LEVEL_DEFAULT, None) self.conn.isolation_level = ext.ISOLATION_LEVEL_DEFAULT self.assertEqual(self.conn.isolation_level, None) cur.execute("SHOW transaction_isolation;") isol = cur.fetchone()[0] cur.execute("SHOW default_transaction_isolation;") self.assertEqual(cur.fetchone()[0], isol)
def test_setattr_isolation_level_str(self): cur = self.conn.cursor() self.conn.isolation_level = "serializable" self.assertEqual(self.conn.isolation_level, ext.ISOLATION_LEVEL_SERIALIZABLE) cur.execute("SHOW transaction_isolation;") self.assertEqual(cur.fetchone()[0], 'serializable') self.conn.rollback() self.conn.isolation_level = "repeatable read" cur.execute("SHOW transaction_isolation;") if self.conn.server_version > 80000: self.assertEqual(self.conn.isolation_level, ext.ISOLATION_LEVEL_REPEATABLE_READ) self.assertEqual(cur.fetchone()[0], 'repeatable read') else: self.assertEqual(self.conn.isolation_level, ext.ISOLATION_LEVEL_SERIALIZABLE) self.assertEqual(cur.fetchone()[0], 'serializable') self.conn.rollback() self.conn.isolation_level = "read committed" self.assertEqual(self.conn.isolation_level, ext.ISOLATION_LEVEL_READ_COMMITTED) cur.execute("SHOW transaction_isolation;") self.assertEqual(cur.fetchone()[0], 'read committed') self.conn.rollback() self.conn.isolation_level = "read uncommitted" cur.execute("SHOW transaction_isolation;") if self.conn.server_version > 80000: self.assertEqual(self.conn.isolation_level, ext.ISOLATION_LEVEL_READ_UNCOMMITTED) self.assertEqual(cur.fetchone()[0], 'read uncommitted') else: self.assertEqual(self.conn.isolation_level, ext.ISOLATION_LEVEL_READ_COMMITTED) self.assertEqual(cur.fetchone()[0], 'read committed') self.conn.rollback() self.conn.isolation_level = "default" self.assertEqual(self.conn.isolation_level, None) cur.execute("SHOW transaction_isolation;") isol = cur.fetchone()[0] cur.execute("SHOW default_transaction_isolation;") self.assertEqual(cur.fetchone()[0], isol)
