python类ISOLATION_LEVEL_REPEATABLE_READ的实例源码

psycopg2.py 文件源码 项目：Lixiang_zhaoxin 作者: hejaxian 项目源码 文件源码 阅读 19 收藏 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 文件源码 项目：flask 作者: bobohope 项目源码 文件源码 阅读 19 收藏 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 文件源码 项目：flask 作者: bobohope 项目源码 文件源码 阅读 34 收藏 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 文件源码 项目：flask 作者: bobohope 项目源码 文件源码 阅读 19 收藏 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 文件源码 项目：Chorus 作者: DonaldBough 项目源码 文件源码 阅读 18 收藏 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 文件源码 项目：Hawkeye 作者: tozhengxq 项目源码 文件源码 阅读 19 收藏 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 文件源码 项目：Alfred 作者: jkachhadia 项目源码 文件源码 阅读 18 收藏 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 文件源码 项目：ngx_status 作者: YoYoAdorkable 项目源码 文件源码 阅读 18 收藏 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 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0 
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)
test_connection.py 文件源码 项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0 
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)
test_connection.py 文件源码 项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0 
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)
test_connection.py 文件源码 项目：ShelbySearch 作者: Agentscreech 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0 
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)
test_connection.py 文件源码 项目：ShelbySearch 作者: Agentscreech 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0 
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)
test_connection.py 文件源码 项目：nmbs-realtime-feed 作者: datamindedbe 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0 
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)
test_connection.py 文件源码 项目：nmbs-realtime-feed 作者: datamindedbe 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0 
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)
__init__.py 文件源码 项目：maas 作者: maas 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0 
def fix_up_databases(databases):
    """Increase isolation level, use atomic requests.

    Does not modify connections to non-PostgreSQL databases.
    """
    # Remove keys with null values from databases.
    databases.update({
        alias: {
            key: value for key, value in database.items()
            if value is not None
        }
        for alias, database in databases.items()
    })
    # Ensure that transactions are configured correctly.
    from psycopg2.extensions import ISOLATION_LEVEL_REPEATABLE_READ
    for _, database in databases.items():
        engine = database.get("ENGINE")
        if engine == 'django.db.backends.postgresql_psycopg2':
            options = database.setdefault("OPTIONS", {})
            # Explicitly set the transaction isolation level. MAAS needs a
            # particular transaction isolation level, and it enforces it.
            if "isolation_level" in options:
                isolation_level = options["isolation_level"]
                if isolation_level != ISOLATION_LEVEL_REPEATABLE_READ:
                    warnings.warn(
                        "isolation_level is set to %r; overriding to %r."
                        % (isolation_level, ISOLATION_LEVEL_REPEATABLE_READ),
                        RuntimeWarning, 2)
            options["isolation_level"] = ISOLATION_LEVEL_REPEATABLE_READ
            # Enable ATOMIC_REQUESTS: MAAS manages transactions across the
            # whole request/response lifecycle including middleware (Django,
            # in its infinite wisdom, does not). However we enable this
            # setting to ensure that views run within _savepoints_ so that
            # middleware exception handlers that suppress exceptions don't
            # inadvertently allow failed requests to be committed.
            if "ATOMIC_REQUESTS" in database:
                atomic_requests = database["ATOMIC_REQUESTS"]
                if not atomic_requests:
                    warnings.warn(
                        "ATOMIC_REQUESTS is set to %r; overriding to True."
                        % (atomic_requests,), RuntimeWarning, 2)
            database["ATOMIC_REQUESTS"] = True
test_connection.py 文件源码 项目：flask 作者: bobohope 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0 
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)
test_connection.py 文件源码 项目：flask 作者: bobohope 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0 
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)

提
问题

写
面经

写
文章

微信
公众号

扫码关注公众号