Repository: aio-libs/aiopg

Issue:
The default read committed isolation level is the wrong choice
Hello, 

I have a project where I experienced serialization failures on a statement:

```
select 1 from file where (id, group_id) = (%s, %s) for update
```

resulting in errors like:

```
psycopg2.errors.SerializationFailure: could not serialize access due to concurrent update
```

now, I happen to know that the statement above would have orderly blocked in case of concurrency, if run in psycopg or in psql: logged some of the operations of the database and sure enough found:

```
postgres_1    | 2020-07-16 10:30:36.377 UTC [11914] LOG:  statement: BEGIN ISOLATION LEVEL REPEATABLE READ
```

Which led me to find:

https://github.com/aio-libs/aiopg/blob/3fb3256319de766384b4867c7cc6710397bd1a8c//aiopg/cursor.py#L10-L16

this is pretty wrong: the driver is choosing an isolation level that causes side effects, and the behaviour of the program becomes very different from the one it would have in psycopg2.

The right default would be to begin a transaction just with `BEGIN`. Note that, unlike what stated here:

https://github.com/aio-libs/aiopg/blob/3fb3256319de766384b4867c7cc6710397bd1a8c/aiopg/transaction.py#L51-L55

`BEGIN` is not "read committed" behaviour: this is "use the server configured behaviour", which is what selected by `default_transaction_isolation` level, and can be changed at server configuration level, session level.

So, what I think is:

1) The `ReadCommittedCompiler` should actually be implemented as:

```python
class ReadCommittedCompiler(IsolationCompiler):
    name = 'Read committed'

    def begin(self):
        return 'BEGIN ISOLATION LEVEL READ COMMITTED'
```

2) you need a class

```python
class DefaultCompiler(IsolationCompiler):
    name = 'Default'

    def begin(self):
        return 'BEGIN'
```

3) that class should be the default.

```python
 class Cursor: 
     def __init__(self, conn, impl, timeout, echo): 
         # ...
         self._transaction = Transaction(self, IsolationLevel.default)
```

you also probably want to fix #497 to allow the isolation level to be chosen, but at least by default you should leave the server default unaltered.

Discussion:
codecov[bot]: # [Codecov](https://codecov.io/gh/aio-libs/aiopg/pull/770?src=pr&el=h1) Report
> Merging [#770](https://codecov.io/gh/aio-libs/aiopg/pull/770?src=pr&el=desc) (a781cb5) into [master](https://codecov.io/gh/aio-libs/aiopg/commit/529c3f1e6aca0bbaad6e61b1ef525289b432b6a9?el=desc) (529c3f1) will **decrease** coverage by `0.73%`.
> The diff coverage is `78.04%`.

[![Impacted file tree graph](https://codecov.io/gh/aio-libs/aiopg/pull/770/graphs/tree.svg?width=650&height=150&src=pr&token=obxFAtiZvt)](https://codecov.io/gh/aio-libs/aiopg/pull/770?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #770      +/-   ##
==========================================
- Coverage   93.08%   92.34%   -0.74%     
==========================================
  Files          13       13              
  Lines        1532     1555      +23     
  Branches      172      174       +2     
==========================================
+ Hits         1426     1436      +10     
- Misses         78       89      +11     
- Partials       28       30       +2     
```


| [Impacted Files](https://codecov.io/gh/aio-libs/aiopg/pull/770?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [aiopg/transaction.py](https://codecov.io/gh/aio-libs/aiopg/pull/770/diff?src=pr&el=tree#diff-YWlvcGcvdHJhbnNhY3Rpb24ucHk=) | `93.61% <77.50%> (-6.39%)` | :arrow_down: |
| [aiopg/cursor.py](https://codecov.io/gh/aio-libs/aiopg/pull/770/diff?src=pr&el=tree#diff-YWlvcGcvY3Vyc29yLnB5) | `100.00% <100.00%> (ø)` | |
| [aiopg/utils.py](https://codecov.io/gh/aio-libs/aiopg/pull/770/diff?src=pr&el=tree#diff-YWlvcGcvdXRpbHMucHk=) | `78.66% <0.00%> (-2.67%)` | :arrow_down: |
| [aiopg/pool.py](https://codecov.io/gh/aio-libs/aiopg/pull/770/diff?src=pr&el=tree#diff-YWlvcGcvcG9vbC5weQ==) | `100.00% <0.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/aio-libs/aiopg/pull/770?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/aio-libs/aiopg/pull/770?src=pr&el=footer). Last update [529c3f1...a781cb5](https://codecov.io/gh/aio-libs/aiopg/pull/770?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/aiopg/connection.py b/aiopg/connection.py
index cdde843..0cc1c3e 100755
--- a/aiopg/connection.py
+++ b/aiopg/connection.py
@@ -199,7 +199,8 @@ class Connection:
         return self._conn.isexecuting()
 
     def cursor(self, name=None, cursor_factory=None,
-               scrollable=None, withhold=False, timeout=None):
+               scrollable=None, withhold=False, timeout=None,
+               isolation_level=None):
         """A coroutine that returns a new cursor object using the connection.
 
         *cursor_factory* argument can be used to create non-standard
@@ -215,11 +216,12 @@ class Connection:
         self._last_usage = self._loop.time()
         coro = self._cursor(name=name, cursor_factory=cursor_factory,
                             scrollable=scrollable, withhold=withhold,
-                            timeout=timeout)
+                            timeout=timeout, isolation_level=isolation_level)
         return _ContextManager(coro)
 
     async def _cursor(self, name=None, cursor_factory=None,
-                      scrollable=None, withhold=False, timeout=None):
+                      scrollable=None, withhold=False, timeout=None,
+                      isolation_level=None):
 
         if not self.closed_cursor:
             warnings.warn(('You can only have one cursor per connection. '
@@ -235,7 +237,9 @@ class Connection:
                                        cursor_factory=cursor_factory,
                                        scrollable=scrollable,
                                        withhold=withhold)
-        self._cursor_instance = Cursor(self, impl, timeout, self._echo)
+        self._cursor_instance = Cursor(
+            self, impl, timeout, self._echo, isolation_level
+        )
         return self._cursor_instance
 
     async def _cursor_impl(self, name=None, cursor_factory=None,
diff --git a/aiopg/cursor.py b/aiopg/cursor.py
index 74d3940..ac8ad44 100644
--- a/aiopg/cursor.py
+++ b/aiopg/cursor.py
@@ -8,12 +8,14 @@ from .utils import _TransactionBeginContextManager
 
 
 class Cursor:
-    def __init__(self, conn, impl, timeout, echo):
+    def __init__(self, conn, impl, timeout, echo, isolation_level):
         self._conn = conn
         self._impl = impl
         self._timeout = timeout
         self._echo = echo
-        self._transaction = Transaction(self, IsolationLevel.repeatable_read)
+        self._transaction = Transaction(
+            self, isolation_level or IsolationLevel.default
+        )
 
     @property
     def echo(self):
diff --git a/aiopg/transaction.py b/aiopg/transaction.py
index ecd22ec..16bea52 100644
--- a/aiopg/transaction.py
+++ b/aiopg/transaction.py
@@ -1,7 +1,7 @@
 import enum
 import uuid
 import warnings
-from abc import ABC, abstractmethod
+from abc import ABC
 
 import psycopg2
 
@@ -11,19 +11,16 @@ __all__ = ('IsolationLevel', 'Transaction')
 
 
 class IsolationCompiler(ABC):
-    name = ''
+    __slots__ = ('_isolation_level', '_readonly', '_deferrable')
 
-    __slots__ = ('_readonly', '_deferrable')
-
-    def __init__(self, readonly, deferrable):
+    def __init__(self, isolation_level, readonly, deferrable):
+        self._isolation_level = isolation_level
         self._readonly = readonly
         self._deferrable = deferrable
-        self._check_readonly_deferrable()
 
-    def _check_readonly_deferrable(self):
-        available = self._readonly or self._deferrable
-        if not isinstance(self, SerializableCompiler) and available:
-            raise ValueError('Is only available for serializable transactions')
+    @property
+    def name(self):
+        return self._isolation_level
 
     def savepoint(self, unique_id):
         return 'SAVEPOINT {}'.format(unique_id)
@@ -40,47 +37,62 @@ class IsolationCompiler(ABC):
     def rollback(self):
         return 'ROLLBACK'
 
-    @abstractmethod
     def begin(self):
-        raise NotImplementedError("Please Implement this method")
+        query = 'BEGIN'
+        if self._isolation_level is not None:
+            query += (
+                ' ISOLATION LEVEL {}'.format(self._isolation_level.upper())
+            )
+
+        if self._readonly:
+            query += ' READ ONLY'
+
+        if self._deferrable:
+            query += ' DEFERRABLE'
+
+        return query
 
     def __repr__(self):
         return self.name
 
 
 class ReadCommittedCompiler(IsolationCompiler):
-    name = 'Read committed'
+    __slots__ = ()
 
-    def begin(self):
-        return 'BEGIN'
+    def __init__(self, readonly, deferrable):
+        super().__init__('Read committed', readonly, deferrable)
 
 
 class RepeatableReadCompiler(IsolationCompiler):
-    name = 'Repeatable read'
+    __slots__ = ()
 
-    def begin(self):
-        return 'BEGIN ISOLATION LEVEL REPEATABLE READ'
+    def __init__(self, readonly, deferrable):
+        super().__init__('Repeatable read', readonly, deferrable)
 
 
 class SerializableCompiler(IsolationCompiler):
-    name = 'Serializable'
+    __slots__ = ()
 
-    def begin(self):
-        query = 'BEGIN ISOLATION LEVEL SERIALIZABLE'
+    def __init__(self, readonly, deferrable):
+        super().__init__('Serializable', readonly, deferrable)
 
-        if self._readonly:
-            query += ' READ ONLY'
 
-        if self._deferrable:
-            query += ' DEFERRABLE'
+class DefaultCompiler(IsolationCompiler):
+    __slots__ = ()
 
-        return query
+    def __init__(self, readonly, deferrable):
+        super().__init__(None, readonly, deferrable)
+
+    @property
+    def name(self):
+        return 'Default'
 
 
 class IsolationLevel(enum.Enum):
     serializable = SerializableCompiler
     repeatable_read = RepeatableReadCompiler
     read_committed = ReadCommittedCompiler
+    default = DefaultCompiler
 
     def __call__(self, readonly, deferrable):
         return self.value(readonly, deferrable)
@@ -173,7 +185,7 @@ class Transaction:
                 ResourceWarning)
 
     async def __aenter__(self):
-        return (await self.begin())
+        return await self.begin()
 
     async def __aexit__(self, exc_type, exc, tb):
         if exc_type is not None:


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index e6b3bd2..c88a209 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -463,6 +463,6 @@ class TcpProxy:
         self.connections.add(client_writer)
 
         await asyncio.wait([
-            self._pipe(server_reader, client_writer),
-            self._pipe(client_reader, server_writer),
+            asyncio.ensure_future(self._pipe(server_reader, client_writer)),
+            asyncio.ensure_future(self._pipe(client_reader, server_writer)),
         ])
diff --git a/tests/test_async_transaction.py b/tests/test_async_transaction.py
index 6ad4cec..abe4e82 100644
--- a/tests/test_async_transaction.py
+++ b/tests/test_async_transaction.py
@@ -17,16 +17,20 @@ def engine(make_connection, loop):
     return loop.run_until_complete(start())
 
 
-@pytest.mark.parametrize('isolation_level,readonly,deferrable', [
-    (IsolationLevel.read_committed, False, False),
-    (IsolationLevel.repeatable_read, False, False),
-    (IsolationLevel.serializable, False, False),
-    (IsolationLevel.serializable, False, True),
+@pytest.mark.parametrize('isolation_level', [
+    IsolationLevel.default,
+    IsolationLevel.read_committed,
+    IsolationLevel.repeatable_read,
+    IsolationLevel.serializable,
 ])
-async def test_transaction(engine, isolation_level, readonly, deferrable):
+@pytest.mark.parametrize('deferrable', [
+    False,
+    True,
+])
+async def test_transaction(engine, isolation_level, deferrable):
     async with engine.cursor() as cur:
         async with Transaction(cur, isolation_level,
-                               readonly=readonly, deferrable=deferrable):
+                               readonly=False, deferrable=deferrable):
             await cur.execute("insert into tbl values(1, 'data')")
             await cur.execute('select * from tbl where id = 1')
             row = await cur.fetchone()
@@ -35,18 +39,40 @@ async def test_transaction(engine, isolation_level, readonly, deferrable):
             assert row[1] == 'data'
 
 
-async def test_transaction_readonly_insert(engine):
+@pytest.mark.parametrize('isolation_level', [
+    IsolationLevel.default,
+    IsolationLevel.read_committed,
+    IsolationLevel.repeatable_read,
+    IsolationLevel.serializable,
+])
+@pytest.mark.parametrize('deferrable', [
+    False,
+    True,
+])
+async def test_transaction_readonly_insert(
+    engine, isolation_level, deferrable
+):
     async with engine.cursor() as cur:
-        async with Transaction(cur, IsolationLevel.serializable,
-                               readonly=True):
+        async with Transaction(cur, isolation_level,
+                               readonly=True, deferrable=deferrable):
             with pytest.raises(psycopg2.InternalError):
                 await cur.execute("insert into tbl values(1, 'data')")
 
 
-async def test_transaction_readonly(engine):
+@pytest.mark.parametrize('isolation_level', [
+    IsolationLevel.default,
+    IsolationLevel.read_committed,
+    IsolationLevel.repeatable_read,
+    IsolationLevel.serializable,
+])
+@pytest.mark.parametrize('deferrable', [
+    False,
+    True,
+])
+async def test_transaction_readonly(engine, isolation_level, deferrable):
     async with engine.cursor() as cur:
-        async with Transaction(cur, IsolationLevel.serializable,
-                               readonly=True):
+        async with Transaction(cur, isolation_level,
+                               readonly=True, deferrable=deferrable):
             await cur.execute('select * from tbl where id = 22')
             row = await cur.fetchone()
 
diff --git a/tests/test_cursor.py b/tests/test_cursor.py
index eca8ff5..6d98d37 100644
--- a/tests/test_cursor.py
+++ b/tests/test_cursor.py
@@ -5,7 +5,9 @@ import psycopg2
 import psycopg2.tz
 import pytest
 
+from aiopg import IsolationLevel
 from aiopg.connection import TIMEOUT
+from aiopg.transaction import ReadCommittedCompiler
 
 
 @pytest.fixture
@@ -304,6 +306,12 @@ async def test_echo_false(connect):
     assert not cur.echo
 
 
+async def test_isolation_level(connect):
+    conn = await connect()
+    cur = await conn.cursor(isolation_level=IsolationLevel.read_committed)
+    assert isinstance(cur._transaction._isolation, ReadCommittedCompiler)
+
+
 async def test_iter(connect):
     conn = await connect()
     cur = await conn.cursor()
diff --git a/tests/test_isolation_level.py b/tests/test_isolation_level.py
new file mode 100644
index 0000000..f1714c1
--- /dev/null
+++ b/tests/test_isolation_level.py
@@ -0,0 +1,68 @@
+import pytest
+
+from aiopg import IsolationLevel
+
+
+@pytest.mark.parametrize('isolation_level, name', [
+    (IsolationLevel.default, 'Default'),
+    (IsolationLevel.read_committed, 'Read committed'),
+    (IsolationLevel.repeatable_read, 'Repeatable read'),
+    (IsolationLevel.serializable, 'Serializable'),
+])
+def test_isolation_level_name(isolation_level, name):
+    assert isolation_level(False, False).name == name
+
+
+@pytest.mark.parametrize(
+    'isolation_level, readonly, deferred, expected_begin',
+    [
+        (IsolationLevel.default, False, False, 'BEGIN'),
+        (IsolationLevel.default, True, False, 'BEGIN READ ONLY'),
+        (
+                IsolationLevel.read_committed,
+                False,
+                False,
+                'BEGIN ISOLATION LEVEL READ COMMITTED'
+        ),
+        (
+                IsolationLevel.read_committed,
+                True,
+                False,
+                'BEGIN ISOLATION LEVEL READ COMMITTED READ ONLY'
+        ),
+        (
+                IsolationLevel.repeatable_read,
+                False,
+                False,
+                'BEGIN ISOLATION LEVEL REPEATABLE READ'
+        ),
+        (
+                IsolationLevel.repeatable_read,
+                True,
+                False,
+                'BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY'
+        ),
+        (
+                IsolationLevel.serializable,
+                False,
+                False,
+                'BEGIN ISOLATION LEVEL SERIALIZABLE'
+        ),
+        (
+                IsolationLevel.serializable,
+                True,
+                False,
+                'BEGIN ISOLATION LEVEL SERIALIZABLE READ ONLY'
+        ),
+        (
+                IsolationLevel.serializable,
+                True,
+                True,
+                'BEGIN ISOLATION LEVEL SERIALIZABLE READ ONLY DEFERRABLE'
+        ),
+    ]
+)
+def test_isolation_level_begin(
+    isolation_level, readonly, deferred, expected_begin
+):
+    assert isolation_level(readonly, deferred).begin() == expected_begin
diff --git a/tests/test_transaction.py b/tests/test_transaction.py
index 1be68c4..985a1e6 100644
--- a/tests/test_transaction.py
+++ b/tests/test_transaction.py
@@ -2,7 +2,6 @@ import psycopg2
 import pytest
 
 from aiopg import IsolationLevel, Transaction
-from aiopg.transaction import IsolationCompiler
 
 
 @pytest.fixture
@@ -20,17 +19,20 @@ def engine(make_engine, loop):
     return loop.run_until_complete(start())
 
 
-@pytest.mark.parametrize('isolation_level,readonly,deferrable', [
-    (IsolationLevel.read_committed, False, False),
-    (IsolationLevel.repeatable_read, False, False),
-    (IsolationLevel.serializable, False, False),
-    (IsolationLevel.serializable, False, True),
+@pytest.mark.parametrize('isolation_level', [
+    IsolationLevel.default,
+    IsolationLevel.read_committed,
+    IsolationLevel.repeatable_read,
+    IsolationLevel.serializable,
 ])
-async def test_transaction_oldstyle(engine, isolation_level, readonly,
-                                    deferrable):
+@pytest.mark.parametrize('deferrable', [
+    False,
+    True,
+])
+async def test_transaction_oldstyle(engine, isolation_level, deferrable):
     async with engine.acquire() as cur:
         tr = Transaction(cur, isolation_level,
-                         readonly=readonly, deferrable=deferrable)
+                         readonly=False, deferrable=deferrable)
         await tr.begin()
         await cur.execute("insert into tbl values(1, 'data')")
         resp = await cur.execute('select * from tbl where id = 1')
@@ -114,22 +116,6 @@ async def test_transaction_fail_oldstyle(engine, fn):
             await fn(cur)
 
 
-def test_transaction_value_error():
-    with pytest.raises(ValueError):
-        Transaction(None, IsolationLevel.read_committed, readonly=True)
-
-
-def test_transaction_isolation_implemented():
-    class IsolationCompilerTest(IsolationCompiler):
-        def begin(self):
-            return super().begin()
-
-    tr = IsolationCompilerTest(False, False)
-
-    with pytest.raises(NotImplementedError):
-        tr.begin()
-
-
 async def test_transaction_finalization_warning(engine, monkeypatch):
     async with engine.acquire() as cur:
         tr = Transaction(cur, IsolationLevel.read_committed)