optimize() and rebuild_fts() now commit their changes

Both ran their INSERT INTO fts(fts) statements via a bare execute()
with no commit, leaving the connection inside an open implicit
transaction - the FTS operation and all subsequent writes were then
silently rolled back when the connection closed. Both are now
wrapped in db.atomic(), matching delete_where() and the other
write operations.

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

sqlite_utils/db.py

@@ -2769,11 +2769,12 @@ class Table(Queryable):
         if fts_table is None:
             # Assume this is itself an FTS table
             fts_table = self.name
-        self.db.execute(
-            "INSERT INTO {table}({table}) VALUES('rebuild');".format(
-                table=quote_identifier(fts_table)
+        with self.db.atomic():
+            self.db.execute(
+                "INSERT INTO {table}({table}) VALUES('rebuild');".format(
+                    table=quote_identifier(fts_table)
+                )
             )
-        )
         return self
 
     def detect_fts(self) -> Optional[str]:
@@ -2805,9 +2806,10 @@ class Table(Queryable):
         "Run the ``optimize`` operation against the associated full-text search index table."
         fts_table = self.detect_fts()
         if fts_table is not None:
-            self.db.execute("""
-                INSERT INTO {table} ({table}) VALUES ("optimize");
-            """.strip().format(table=quote_identifier(fts_table)))
+            with self.db.atomic():
+                self.db.execute("""
+                    INSERT INTO {table} ({table}) VALUES ("optimize");
+                """.strip().format(table=quote_identifier(fts_table)))
         return self
 
     def search_sql(

tests/test_fts.py

@@ -336,6 +336,24 @@ def test_rebuild_fts(fresh_db):
     assert len(rows2) == 2
 
 
+@pytest.mark.parametrize("method", ["optimize", "rebuild_fts"])
+def test_optimize_and_rebuild_fts_commit(tmpdir, method):
+    path = str(tmpdir / "test.db")
+    db = Database(path)
+    table = db["searchable"]
+    table.insert(search_records[0])
+    table.enable_fts(["text", "country"])
+    getattr(table, method)()
+    # The connection must not be left inside an open transaction,
+    # otherwise this and all subsequent writes are lost on close
+    assert not db.conn.in_transaction
+    table.insert(search_records[1])
+    db.close()
+    db2 = Database(path)
+    assert db2["searchable"].count == 2
+    db2.close()
+
+
 @pytest.mark.parametrize("invalid_table", ["does_not_exist", "not_searchable"])
 def test_rebuild_fts_invalid(fresh_db, invalid_table):
     fresh_db["not_searchable"].insert({"foo": "bar"})