Run each migration inside a transaction

Migrations.apply() now wraps each migration function and its
_sqlite_migrations tracking row in db.atomic(), so they commit
together: a failing migration rolls back cleanly, is not recorded,
and stays pending, eliminating the double-apply hazard where a
partially-failed migration left committed side effects behind.

Migrations that cannot run inside a transaction (VACUUM, journal
mode changes, manual transaction management) can opt out by
registering with @migrations(transactional=False).

Also fixed the test fixtures which used db.query() for INSERT
statements - query() is a lazy generator, so those inserts never
actually executed.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UnLnhsH25Nnv7LHhekUfPd

sqlite_utils/migrations.py

@@ -14,6 +14,7 @@ class Migrations:
     class _Migration:
         name: str
         fn: Callable
+        transactional: bool = True
 
     @dataclass
     class _AppliedMigration:
@@ -27,15 +28,22 @@ class Migrations:
         self.name = name
         self._migrations: list[Migrations._Migration] = []
 
-    def __call__(self, *, name: str | None = None) -> Callable:
+    def __call__(
+        self, *, name: str | None = None, transactional: bool = True
+    ) -> Callable:
         """
         :param name: The name to use for this migration - if not provided,
           the name of the function will be used.
+        :param transactional: If ``True`` (the default) the migration and the
+          record of it having been applied are wrapped in a transaction, which
+          will be rolled back if the migration raises an exception. Pass
+          ``False`` for migrations that cannot run inside a transaction, for
+          example those that execute ``VACUUM``.
         """
 
         def inner(func: Callable) -> Callable:
             self._migrations.append(
-                self._Migration(name or getattr(func, "__name__"), func)
+                self._Migration(name or getattr(func, "__name__"), func, transactional)
             )
             return func
 
@@ -73,6 +81,12 @@ class Migrations:
     def apply(self, db: "Database", *, stop_before: str | Iterable[str] | None = None):
         """
         Apply any pending migrations to the database.
+
+        Each migration runs inside a transaction, together with the record of
+        it having been applied - if the migration raises an exception its
+        changes are rolled back, no record is written and the migration stays
+        pending. Migrations registered with ``transactional=False`` run
+        outside of a transaction.
         """
         self.ensure_migrations_table(db)
         if stop_before is None:
@@ -85,14 +99,22 @@ class Migrations:
             name = migration.name
             if name in stop_before_names:
                 return
-            migration.fn(db)
-            _table(db, self.migrations_table).insert(
-                {
-                    "migration_set": self.name,
-                    "name": name,
-                    "applied_at": str(datetime.datetime.now(datetime.timezone.utc)),
-                }
-            )
+            if migration.transactional:
+                with db.atomic():
+                    migration.fn(db)
+                    self._record_applied(db, name)
+            else:
+                migration.fn(db)
+                self._record_applied(db, name)
+
+    def _record_applied(self, db: "Database", name: str):
+        _table(db, self.migrations_table).insert(
+            {
+                "migration_set": self.name,
+                "name": name,
+                "applied_at": str(datetime.datetime.now(datetime.timezone.utc)),
+            }
+        )
 
     def ensure_migrations_table(self, db: "Database"):
         """