feat(client): add transaction isolation level

Author: jonathanblade

databases/sync_tests/test_transactions.py

@@ -201,3 +201,124 @@ def test_transaction_already_closed(client: Prisma) -> None:
         transaction.user.delete_many()
 
     assert exc.match('Transaction already closed')
+
+
+@pytest.mark.skipif(CURRENT_DATABASE in ['cockroachdb', 'sqlite'], reason='Not available')
+def test_read_uncommited_isolation_level(client: Prisma) -> None:
+    """A transaction isolation level is set to `READ_UNCOMMITED`"""
+    client2 = Prisma()
+    client2.connect()
+
+    user = client.user.create(data={'name': 'Robert'})
+
+    with client.tx(isolation_level=prisma.TransactionIsolationLevel.READ_UNCOMMITED) as tx1:
+        tx1_user = tx1.user.find_first_or_raise(where={'id': user.id})
+        tx1_count = tx1.user.count()
+
+        with client2.tx() as tx2:
+            tx2.user.update(data={'name': 'Tegan'}, where={'id': user.id})
+            tx2.user.create(data={'name': 'Bobby'})
+
+            dirty_user = tx1.user.find_first_or_raise(where={'id': user.id})
+
+        non_repeatable_user = tx1.user.find_first_or_raise(where={'id': user.id})
+        phantom_count = tx1.user.count()
+
+    # Have dirty read
+    assert tx1_user.name != dirty_user.name
+    # Have non-repeatable read
+    assert tx1_user.name != non_repeatable_user.name
+    # Have phantom read
+    assert tx1_count != phantom_count
+
+
+@pytest.mark.skipif(CURRENT_DATABASE in ['cockroachdb', 'sqlite'], reason='Not available')
+def test_read_commited_isolation_level(client: Prisma) -> None:
+    """A transaction isolation level is set to `READ_COMMITED`"""
+    client2 = Prisma()
+    client2.connect()
+
+    user = client.user.create(data={'name': 'Robert'})
+
+    with client.tx(isolation_level=prisma.TransactionIsolationLevel.READ_COMMITED) as tx1:
+        tx1_user = tx1.user.find_first_or_raise(where={'id': user.id})
+        tx1_count = tx1.user.count()
+
+        with client2.tx() as tx2:
+            tx2.user.update(data={'name': 'Tegan'}, where={'id': user.id})
+            tx2.user.create(data={'name': 'Bobby'})
+
+            dirty_user = tx1.user.find_first_or_raise(where={'id': user.id})
+
+        non_repeatable_user = tx1.user.find_first_or_raise(where={'id': user.id})
+        phantom_count = tx1.user.count()
+
+    # No dirty read
+    assert tx1_user.name == dirty_user.name
+    # Have non-repeatable read
+    assert tx1_user.name != non_repeatable_user.name
+    # Have phantom read
+    assert tx1_count != phantom_count
+
+
+@pytest.mark.skipif(CURRENT_DATABASE in ['cockroachdb', 'sqlite'], reason='Not available')
+def test_repeatable_read_isolation_level(client: Prisma) -> None:
+    """A transaction isolation level is set to `REPEATABLE_READ`"""
+    client2 = Prisma()
+    client2.connect()
+
+    user = client.user.create(data={'name': 'Robert'})
+
+    with client.tx(isolation_level=prisma.TransactionIsolationLevel.REPEATABLE_READ) as tx1:
+        tx1_user = tx1.user.find_first_or_raise(where={'id': user.id})
+        tx1_count = tx1.user.count()
+
+        with client2.tx() as tx2:
+            tx2.user.update(data={'name': 'Tegan'}, where={'id': user.id})
+            tx2.user.create(data={'name': 'Bobby'})
+
+            dirty_user = tx1.user.find_first_or_raise(where={'id': user.id})
+
+        non_repeatable_user = tx1.user.find_first_or_raise(where={'id': user.id})
+        phantom_count = tx1.user.count()
+
+    # No dirty read
+    assert tx1_user.name == dirty_user.name
+    # No non-repeatable read
+    assert tx1_user.name == non_repeatable_user.name
+    # Have phantom read
+    assert tx1_count != phantom_count
+
+
+@pytest.mark.skipif(True, reason='Available for SQL Server only')
+def test_snapshot_isolation_level() -> None:
+    """A transaction isolation level is set to `SNAPSHOT`"""
+    raise NotImplementedError
+
+
+def test_serializable_isolation_level(client: Prisma) -> None:
+    """A transaction isolation level is set to `SERIALIZABLE`"""
+    client2 = Prisma()
+    client2.connect()
+
+    user = client.user.create(data={'name': 'Robert'})
+
+    with client.tx(isolation_level=prisma.TransactionIsolationLevel.SERIALIZABLE) as tx1:
+        tx1_user = tx1.user.find_first_or_raise(where={'id': user.id})
+        tx1_count = tx1.user.count()
+
+        with client2.tx() as tx2:
+            tx2.user.update(data={'name': 'Tegan'}, where={'id': user.id})
+            tx2.user.create(data={'name': 'Bobby'})
+
+            dirty_user = tx1.user.find_first_or_raise(where={'id': user.id})
+
+        non_repeatable_user = tx1.user.find_first_or_raise(where={'id': user.id})
+        phantom_count = tx1.user.count()
+
+    # No dirty read
+    assert tx1_user.name == dirty_user.name
+    # No non-repeatable read
+    assert tx1_user.name == non_repeatable_user.name
+    # No phantom read
+    assert tx1_count == phantom_count

databases/tests/test_transactions.py

@@ -212,3 +212,129 @@ async def test_transaction_already_closed(client: Prisma) -> None:
         await transaction.user.delete_many()
 
     assert exc.match('Transaction already closed')
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(CURRENT_DATABASE in ['cockroachdb', 'sqlite'], reason='Not available')
+async def test_read_uncommited_isolation_level(client: Prisma) -> None:
+    """A transaction isolation level is set to `READ_UNCOMMITED`"""
+    client2 = Prisma()
+    await client2.connect()
+
+    user = await client.user.create(data={'name': 'Robert'})
+
+    async with client.tx(isolation_level=prisma.TransactionIsolationLevel.READ_UNCOMMITED) as tx1:
+        tx1_user = await tx1.user.find_first_or_raise(where={'id': user.id})
+        tx1_count = await tx1.user.count()
+
+        async with client2.tx() as tx2:
+            await tx2.user.update(data={'name': 'Tegan'}, where={'id': user.id})
+            await tx2.user.create(data={'name': 'Bobby'})
+
+            dirty_user = await tx1.user.find_first_or_raise(where={'id': user.id})
+
+        non_repeatable_user = await tx1.user.find_first_or_raise(where={'id': user.id})
+        phantom_count = await tx1.user.count()
+
+    # Have dirty read
+    assert tx1_user.name != dirty_user.name
+    # Have non-repeatable read
+    assert tx1_user.name != non_repeatable_user.name
+    # Have phantom read
+    assert tx1_count != phantom_count
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(CURRENT_DATABASE in ['cockroachdb', 'sqlite'], reason='Not available')
+async def test_read_commited_isolation_level(client: Prisma) -> None:
+    """A transaction isolation level is set to `READ_COMMITED`"""
+    client2 = Prisma()
+    await client2.connect()
+
+    user = await client.user.create(data={'name': 'Robert'})
+
+    async with client.tx(isolation_level=prisma.TransactionIsolationLevel.READ_COMMITED) as tx1:
+        tx1_user = await tx1.user.find_first_or_raise(where={'id': user.id})
+        tx1_count = await tx1.user.count()
+
+        async with client2.tx() as tx2:
+            await tx2.user.update(data={'name': 'Tegan'}, where={'id': user.id})
+            await tx2.user.create(data={'name': 'Bobby'})
+
+            dirty_user = await tx1.user.find_first_or_raise(where={'id': user.id})
+
+        non_repeatable_user = await tx1.user.find_first_or_raise(where={'id': user.id})
+        phantom_count = await tx1.user.count()
+
+    # No dirty read
+    assert tx1_user.name == dirty_user.name
+    # Have non-repeatable read
+    assert tx1_user.name != non_repeatable_user.name
+    # Have phantom read
+    assert tx1_count != phantom_count
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(CURRENT_DATABASE in ['cockroachdb', 'sqlite'], reason='Not available')
+async def test_repeatable_read_isolation_level(client: Prisma) -> None:
+    """A transaction isolation level is set to `REPEATABLE_READ`"""
+    client2 = Prisma()
+    await client2.connect()
+
+    user = await client.user.create(data={'name': 'Robert'})
+
+    async with client.tx(isolation_level=prisma.TransactionIsolationLevel.REPEATABLE_READ) as tx1:
+        tx1_user = await tx1.user.find_first_or_raise(where={'id': user.id})
+        tx1_count = await tx1.user.count()
+
+        async with client2.tx() as tx2:
+            await tx2.user.update(data={'name': 'Tegan'}, where={'id': user.id})
+            await tx2.user.create(data={'name': 'Bobby'})
+
+            dirty_user = await tx1.user.find_first_or_raise(where={'id': user.id})
+
+        non_repeatable_user = await tx1.user.find_first_or_raise(where={'id': user.id})
+        phantom_count = await tx1.user.count()
+
+    # No dirty read
+    assert tx1_user.name == dirty_user.name
+    # No non-repeatable read
+    assert tx1_user.name == non_repeatable_user.name
+    # Have phantom read
+    assert tx1_count != phantom_count
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(True, reason='Available for SQL Server only')
+async def test_snapshot_isolation_level() -> None:
+    """A transaction isolation level is set to `SNAPSHOT`"""
+    raise NotImplementedError
+
+
+@pytest.mark.asyncio
+async def test_serializable_isolation_level(client: Prisma) -> None:
+    """A transaction isolation level is set to `SERIALIZABLE`"""
+    client2 = Prisma()
+    await client2.connect()
+
+    user = await client.user.create(data={'name': 'Robert'})
+
+    async with client.tx(isolation_level=prisma.TransactionIsolationLevel.SERIALIZABLE) as tx1:
+        tx1_user = await tx1.user.find_first_or_raise(where={'id': user.id})
+        tx1_count = await tx1.user.count()
+
+        async with client2.tx() as tx2:
+            await tx2.user.update(data={'name': 'Tegan'}, where={'id': user.id})
+            await tx2.user.create(data={'name': 'Bobby'})
+
+            dirty_user = await tx1.user.find_first_or_raise(where={'id': user.id})
+
+        non_repeatable_user = await tx1.user.find_first_or_raise(where={'id': user.id})
+        phantom_count = await tx1.user.count()
+
+    # No dirty read
+    assert tx1_user.name == dirty_user.name
+    # No non-repeatable read
+    assert tx1_user.name == non_repeatable_user.name
+    # No phantom read
+    assert tx1_count == phantom_count

src/prisma/generator/templates/client.py.jinja

@@ -5,6 +5,7 @@
 import warnings
 import logging
 from datetime import timedelta
+from enum import Enum
 from pathlib import Path
 from types import TracebackType
 
@@ -39,6 +40,7 @@ __all__ = (
     'Batch',
     'Prisma',
     'Client',
+    'TransactionIsolationLevel',
     'load_env',
     'register',
     'get_client',
@@ -323,6 +325,7 @@ class Prisma:
     def tx(
         self,
         *,
+        isolation_level: Optional['TransactionIsolationLevel'] = None,
         max_wait: Union[int, timedelta] = DEFAULT_TX_MAX_WAIT,
         timeout: Union[int, timedelta] = DEFAULT_TX_TIMEOUT,
     ) -> 'TransactionManager':
@@ -332,6 +335,9 @@ class Prisma:
         actions within a transaction, queries will be isolated to the Prisma instance and
         will not be commited to the database until the context manager exits.
 
+        By default, Prisma sets the isolation level to the value currently configured in the database. You can modify this
+        default with the `isolation_level` argument (see [supported isolation levels](https://www.prisma.io/docs/orm/prisma-client/queries/transactions#supported-isolation-levels)).
+
         By default, Prisma will wait a maximum of 2 seconds to acquire a transaction from the database. You can modify this
         default with the `max_wait` argument which accepts a value in milliseconds or `datetime.timedelta`.
 
@@ -348,7 +354,7 @@ class Prisma:
 
         In the above example, if the first database call succeeds but the second does not then neither of the records will be created.
         """
-        return TransactionManager(client=self, max_wait=max_wait, timeout=timeout)
+        return TransactionManager(client=self, isolation_level=isolation_level, max_wait=max_wait, timeout=timeout)
 
     def is_transaction(self) -> bool:
         """Returns True if the client is wrapped within a transaction"""
@@ -475,16 +481,34 @@ class Prisma:
         }
 
 
+# See here: https://www.prisma.io/docs/orm/prisma-client/queries/transactions#supported-isolation-levels
+class TransactionIsolationLevel(str, Enum):
+    READ_UNCOMMITED = "ReadUncommitted"
+    READ_COMMITED = "ReadCommitted"
+    REPEATABLE_READ = "RepeatableRead"
+    SNAPSHOT = "Snapshot"
+    SERIALIZABLE = "Serializable"
+
+
 class TransactionManager:
     """Context manager for wrapping a Prisma instance within a transaction.
 
     This should never be created manually, instead it should be used
     through the Prisma.tx() method.
     """
 
-    def __init__(self, *, client: Prisma, max_wait: Union[int, timedelta], timeout: Union[int, timedelta]) -> None:
+    def __init__(
+        self,
+        *,
+        client: Prisma,
+        isolation_level: Optional['TransactionIsolationLevel'],
+        max_wait: Union[int, timedelta],
+        timeout: Union[int, timedelta],
+    ) -> None:
         self.__client = client
 
+        self._isolation_level = isolation_level
+
         if isinstance(max_wait, int):
             message = (
                 'Passing an int as `max_wait` argument is deprecated '
@@ -520,14 +544,14 @@ class TransactionManager:
                 stacklevel=3 if _from_context else 2
             )
 
-        tx_id = {{ maybe_await }}self.__client._engine.start_transaction(
-            content=dumps(
-                {
-                    'timeout': int(self._timeout.total_seconds() * 1000),
-                    'max_wait': int(self._max_wait.total_seconds() * 1000),
-                }
-            ),
-        )
+        content_dict: dict[str, Any] = {
+            'timeout': int(self._timeout.total_seconds() * 1000),
+            'max_wait': int(self._max_wait.total_seconds() * 1000),
+        }
+        if self._isolation_level:
+            content_dict['isolation_level'] = self._isolation_level.value
+
+        tx_id = {{ maybe_await }}self.__client._engine.start_transaction(content=dumps(content_dict))
         self._tx_id = tx_id
         client = self.__client._copy()
         client._tx_id = tx_id