add wrap_in_transaction option (#1069)

piccolo/apps/migrations/auto/migration_manager.py

@@ -130,11 +130,27 @@ def table_class_names(self) -> t.List[str]:
 AsyncFunction = t.Callable[[], t.Coroutine]
 
 
+class SkippedTransaction:
+    async def __aenter__(self):
+        print("Automatic transaction disabled")
+
+    async def __aexit__(self, *args, **kwargs):
+        pass
+
+
 @dataclass
 class MigrationManager:
     """
     Each auto generated migration returns a MigrationManager. It contains
     all of the schema changes that migration wants to make.
+
+    :param wrap_in_transaction:
+        By default, the migration is wrapped in a transaction, so if anything
+        fails, the whole migration will get rolled back. You can disable this
+        behaviour if you want - for example, in a manual migration you might
+        want to create the transaction yourself (perhaps you're using
+        savepoints), or you may want multiple transactions.
+
     """
 
     migration_id: str = ""
@@ -166,6 +182,7 @@ class MigrationManager:
         default_factory=list
     )
     fake: bool = False
+    wrap_in_transaction: bool = True
 
     def add_table(
         self,
@@ -935,7 +952,11 @@ async def run(self, backwards: bool = False):
         if not engine:
             raise Exception("Can't find engine")
 
-        async with engine.transaction():
+        async with (
+            engine.transaction()
+            if self.wrap_in_transaction
+            else SkippedTransaction()
+        ):
             if not self.preview:
                 if direction == "backwards":
                     raw_list = self.raw_backwards

tests/apps/migrations/auto/test_migration_manager.py

@@ -2,7 +2,7 @@
 import random
 import typing as t
 from io import StringIO
-from unittest import TestCase
+from unittest import IsolatedAsyncioTestCase, TestCase
 from unittest.mock import MagicMock, patch
 
 from piccolo.apps.migrations.auto.migration_manager import MigrationManager
@@ -11,6 +11,7 @@
 from piccolo.columns.base import OnDelete, OnUpdate
 from piccolo.columns.column_types import ForeignKey
 from piccolo.conf.apps import AppConfig
+from piccolo.engine import engine_finder
 from piccolo.table import Table, sort_table_classes
 from piccolo.utils.lazy_loader import LazyLoader
 from tests.base import AsyncMock, DBTestCase, engine_is, engines_only
@@ -1052,3 +1053,37 @@ def test_change_table_schema(self):
                 output,
                 '  - 1 [preview forwards]... CREATE SCHEMA IF NOT EXISTS "schema_1"\nALTER TABLE "manager" SET SCHEMA "schema_1"\n',  # noqa: E501
             )
+
+
+class TestWrapInTransaction(IsolatedAsyncioTestCase):
+
+    async def test_enabled(self):
+        """
+        Make sure we can wrap the migration in a transaction if we want to.
+        """
+
+        async def run():
+            db = engine_finder()
+            assert db
+            assert db.transaction_exists() is True
+
+        manager = MigrationManager(wrap_in_transaction=True)
+        manager.add_raw(run)
+
+        await manager.run()
+
+    async def test_disabled(self):
+        """
+        Make sure we can stop the migration being wrapped in a transaction if
+        we want to.
+        """
+
+        async def run():
+            db = engine_finder()
+            assert db
+            assert db.transaction_exists() is False
+
+        manager = MigrationManager(wrap_in_transaction=False)
+        manager.add_raw(run)
+
+        await manager.run()