Add AUTOCOMMIT isolation level support for psycopg2

One can use this to emit statements, which can not be
executed within a transaction (e. g. CREATE DATABASE):

    from sqlalchemy import create_engine

    eng = create_engine('postgresql://test:test@localhost/test')

    conn = eng.connect().execution_options(isolation_level='AUTOCOMMIT')
    conn.execute('CREATE DATABASE test2;')

Fixes issue #2072.
This commit is contained in:
Roman Podolyaka
2013-06-15 15:38:02 +03:00
parent 9d3e220636
commit a1ceae2ed3
2 changed files with 11 additions and 0 deletions
@@ -350,6 +350,7 @@ class PGDialect_psycopg2(PGDialect):
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,
+10
View File
@@ -1890,6 +1890,16 @@ class MiscTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
c = e.connect()
eq_(c.connection.connection.encoding, test_encoding)
@testing.only_on('postgresql+psycopg2', 'psycopg2-specific feature')
@engines.close_open_connections
def test_autocommit_isolation_level(self):
extensions = __import__('psycopg2.extensions').extensions
c = testing.db.connect()
c = c.execution_options(isolation_level='AUTOCOMMIT')
eq_(c.connection.connection.isolation_level,
extensions.ISOLATION_LEVEL_AUTOCOMMIT)
@testing.fails_on('+zxjdbc',
"Can't infer the SQL type to use for an instance "
"of org.python.core.PyObjectDerived.")