fix(adapters): implement new transactions (#62)

Author: emptybutton

src/ttt/infrastructure/adapters/transaction.py

@@ -1,34 +1,56 @@
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from types import TracebackType
 from typing import Self
 
-from sqlalchemy.ext.asyncio import AsyncSession, AsyncSessionTransaction
+from sqlalchemy.ext.asyncio import AsyncSession
 
-from ttt.application.common.ports.transaction import Transaction
-from ttt.entities.tools.assertion import not_none
+from ttt.application.common.ports.transaction import (
+    NotSerializableTransaction,
+    ReadonlyTransaction,
+    SerializableTransaction,
+)
+from ttt.entities.tools.assertion import assert_, not_none
+from ttt.infrastructure.sqlalchemy.serialization import (
+    reraise_serialization_error,
+)
 
 
 @dataclass
-class InPostgresTransaction(Transaction):
+class InPostgresSerializableTransaction(SerializableTransaction):
     _session: AsyncSession
-    _transaction: AsyncSessionTransaction | None = field(
-        init=False,
-        default=None,
-    )
-    _nesting_counter: int = field(
-        init=False,
-        default=0,
-    )
 
     async def __aenter__(self) -> Self:
-        self._nesting_counter += 1
+        assert_(not self._session.in_transaction())
+        await self._session.connection(
+            execution_options={"isolation_level": "SERIALIZABLE"},
+        )
+        return self
 
-        if self._transaction is None:
-            self._transaction = await self._session.begin()
-        elif not self._transaction.is_active:
-            await self._session.rollback()
-            self._transaction = await self._session.begin()
+    async def __aexit__(
+        self,
+        error_type: type[BaseException] | None,
+        error: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None:
+        transaction = not_none(self._session.get_transaction())
+        with reraise_serialization_error():
+            await transaction.__aexit__(error_type, error, traceback)
 
+    async def commit(self) -> None:
+        transaction = not_none(self._session.get_transaction())
+        with reraise_serialization_error():
+            await transaction.commit()
+
+
+@dataclass
+class InPostgresNotSerializableTransaction(NotSerializableTransaction):
+    _session: AsyncSession
+
+    async def __aenter__(self) -> Self:
+        assert_(not self._session.in_transaction())
+        await self._session.connection(
+            execution_options={"isolation_level": "READ COMMITED"},
+        )
         return self
 
     async def __aexit__(
@@ -37,10 +59,29 @@ async def __aexit__(
         error: BaseException | None,
         traceback: TracebackType | None,
     ) -> None:
-        self._nesting_counter -= 1
+        transaction = not_none(self._session.get_transaction())
+        await transaction.__aexit__(error_type, error, traceback)
 
-        if self._nesting_counter == 0:
-            transaction = not_none(self._transaction)
-            await transaction.__aexit__(error_type, error, traceback)
-            self._transaction = None
-            return
+    async def commit(self) -> None:
+        transaction = not_none(self._session.get_transaction())
+        await transaction.commit()
+
+
+@dataclass
+class InPostgresReadonlyTransaction(ReadonlyTransaction):
+    _session: AsyncSession
+
+    async def __aenter__(self) -> Self:
+        assert_(not self._session.in_transaction())
+        options = {"isolation_level": "SERIALIZABLE", "readonly": True}
+        await self._session.connection(execution_options=options)
+        return self
+
+    async def __aexit__(
+        self,
+        error_type: type[BaseException] | None,
+        error: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None:
+        transaction = not_none(self._session.get_transaction())
+        await transaction.__aexit__(error_type, error, traceback)

src/ttt/infrastructure/sqlalchemy/serialization.py

@@ -0,0 +1,18 @@
+from collections.abc import Iterator
+from contextlib import contextmanager
+
+from psycopg.errors import SerializationFailure
+from sqlalchemy.exc import OperationalError
+
+from ttt.application.common.errors.serialization_error import SerializationError
+
+
+@contextmanager
+def reraise_serialization_error() -> Iterator[None]:
+    try:
+        yield
+    except OperationalError as error:
+        if isinstance(error.orig, SerializationFailure):
+            raise SerializationError from error
+
+        raise error from error