sqlalchemy adapter trx begin allow transaction_mode (#498)

* sqlalchemy adapter trx begin allow transaction_mode

* adding tests

aiopg/sa/connection.py

@@ -134,9 +134,19 @@ def info(self):
     def connection(self):
         return self._connection
 
-    def begin(self):
+    def begin(self, isolation_level=None, readonly=False, deferrable=False):
         """Begin a transaction and return a transaction handle.
 
+        isolation_level - The isolation level of the transaction,
+        should be one of 'SERIALIZABLE', 'REPEATABLE READ', 'READ COMMITTED',
+        'READ UNCOMMITTED', default (None) is 'READ COMMITTED'
+
+        readonly - The transaction is read only
+
+        deferrable - The transaction may block when acquiring data before
+        running without overhead of SERLIALIZABLE, has no effect unless
+        transaction is both SERIALIZABLE and readonly
+
         The returned object is an instance of Transaction.  This
         object represents the "scope" of the transaction, which
         completes when either the .rollback or .commit method is
@@ -161,23 +171,31 @@ def begin(self):
           .begin_twophase - use a two phase/XA transaction
 
         """
-        coro = self._begin()
+        coro = self._begin(isolation_level, readonly, deferrable)
         return _TransactionContextManager(coro)
 
     @asyncio.coroutine
-    def _begin(self):
+    def _begin(self, isolation_level, readonly, deferrable):
         if self._transaction is None:
             self._transaction = RootTransaction(self)
-            yield from self._begin_impl()
+            yield from self._begin_impl(isolation_level, readonly, deferrable)
             return self._transaction
         else:
             return Transaction(self, self._transaction)
 
     @asyncio.coroutine
-    def _begin_impl(self):
+    def _begin_impl(self, isolation_level, readonly, deferrable):
+        stmt = 'BEGIN'
+        if isolation_level is not None:
+            stmt += ' ISOLATION LEVEL ' + isolation_level
+        if readonly:
+            stmt += ' READ ONLY'
+        if deferrable:
+            stmt += ' DEFERRABLE'
+
         cur = yield from self._connection.cursor()
         try:
-            yield from cur.execute('BEGIN')
+            yield from cur.execute(stmt)
         finally:
             cur.close()
 
@@ -217,7 +235,7 @@ def begin_nested(self):
     def _begin_nested(self):
         if self._transaction is None:
             self._transaction = RootTransaction(self)
-            yield from self._begin_impl()
+            yield from self._begin_impl(None, False, False)
         else:
             self._transaction = NestedTransaction(self, self._transaction)
             self._transaction._savepoint = yield from self._savepoint_impl()

tests/test_sa_transaction.py

@@ -377,3 +377,60 @@ def test_transactions_sequence(xa_connect):
 
     yield from tr3.commit()
     assert conn._transaction is None
+
+
+@asyncio.coroutine
+def test_transaction_mode(connect):
+    conn = yield from connect()
+
+    yield from conn.execute(tbl.delete())
+
+    tr1 = yield from conn.begin(isolation_level='SERIALIZABLE')
+    yield from conn.execute(tbl.insert().values(name='a'))
+    res1 = yield from conn.scalar(tbl.count())
+    assert 1 == res1
+    yield from tr1.commit()
+
+    tr2 = yield from conn.begin(isolation_level='REPEATABLE READ')
+    yield from conn.execute(tbl.insert().values(name='b'))
+    res2 = yield from conn.scalar(tbl.count())
+    assert 2 == res2
+    yield from tr2.commit()
+
+    tr3 = yield from conn.begin(isolation_level='READ UNCOMMITTED')
+    yield from conn.execute(tbl.insert().values(name='c'))
+    res3 = yield from conn.scalar(tbl.count())
+    assert 3 == res3
+    yield from tr3.commit()
+
+    tr4 = yield from conn.begin(readonly=True)
+    assert tr4 is conn._transaction
+    res1 = yield from conn.scalar(tbl.count())
+    assert 3 == res1
+    yield from tr4.commit()
+
+    tr5 = yield from conn.begin(isolation_level='READ UNCOMMITTED',
+                                readonly=True)
+    res1 = yield from conn.scalar(tbl.count())
+    assert 3 == res1
+    yield from tr5.commit()
+
+    tr6 = yield from conn.begin(deferrable=True)
+    yield from conn.execute(tbl.insert().values(name='f'))
+    res1 = yield from conn.scalar(tbl.count())
+    assert 4 == res1
+    yield from tr6.commit()
+
+    tr7 = yield from conn.begin(isolation_level='REPEATABLE READ',
+                                deferrable=True)
+    yield from conn.execute(tbl.insert().values(name='g'))
+    res1 = yield from conn.scalar(tbl.count())
+    assert 5 == res1
+    yield from tr7.commit()
+
+    tr8 = yield from conn.begin(isolation_level='SERIALIZABLE',
+                                readonly=True, deferrable=True)
+    assert tr8 is conn._transaction
+    res1 = yield from conn.scalar(tbl.count())
+    assert 5 == res1
+    yield from tr8.commit()