From 8fc93c4c68c28038c7d15e595cdc6d13cabccfe1 Mon Sep 17 00:00:00 2001
From: ArockiaRajamanickam <arockia.lyx@gmail.com>
Date: Thu, 30 Jul 2026 19:31:57 +0530
Subject: [PATCH] Detect augmented JSON types when skipping batch migrate CAST

The SQLite batch migration data transfer suppresses the CAST for JSON
columns, since CASTing to JSON in SQLite yields 0. The check used
isinstance(), which does not match a TypeDecorator that augments JSON,
so the CAST was emitted and the column data was replaced with 0.
Compare type affinity instead, as the preceding condition already does.

Fixes: #1120
---
 alembic/ddl/sqlite.py          |  5 ++++-
 docs/build/unreleased/1120.rst | 12 ++++++++++++
 tests/test_batch.py            | 28 ++++++++++++++++++++++++++++
 3 files changed, 44 insertions(+), 1 deletion(-)
 create mode 100644 docs/build/unreleased/1120.rst

diff --git a/alembic/ddl/sqlite.py b/alembic/ddl/sqlite.py
index c260d53fa..1a38d22d4 100644
--- a/alembic/ddl/sqlite.py
+++ b/alembic/ddl/sqlite.py
@@ -182,9 +182,12 @@ def cast_for_batch_migrate(
         existing_transfer: Dict[str, Union[TypeEngine, Cast]],
         new_type: TypeEngine,
     ) -> None:
+        # the JSON check uses the type affinity rather than isinstance() so
+        # that a TypeDecorator which augments JSON is also detected; CASTing
+        # to JSON in SQLite yields 0 and would destroy the data
         if (
             existing.type._type_affinity is not new_type._type_affinity
-            and not isinstance(new_type, JSON)
+            and new_type._type_affinity is not JSON
         ):
             existing_transfer["expr"] = cast(
                 existing_transfer["expr"], new_type
diff --git a/docs/build/unreleased/1120.rst b/docs/build/unreleased/1120.rst
new file mode 100644
index 000000000..79fbb80a4
--- /dev/null
+++ b/docs/build/unreleased/1120.rst
@@ -0,0 +1,12 @@
+.. change::
+    :tags: bug, batch migrations
+    :tickets: 1120
+
+    Fixed data loss in SQLite batch migrations when altering a column to a
+    :class:`~sqlalchemy.types.TypeDecorator` that augments
+    :class:`~sqlalchemy.types.JSON`.  The data transfer step suppresses the
+    ``CAST`` for :class:`~sqlalchemy.types.JSON`, as CASTing to ``JSON`` in
+    SQLite yields ``0``, however the check used ``isinstance()`` and therefore
+    did not match a ``TypeDecorator`` wrapping ``JSON``; every value in the
+    column was replaced with ``0``.  The check now compares type affinity, so
+    augmented types are detected as well.
diff --git a/tests/test_batch.py b/tests/test_batch.py
index 0dfd273b3..cb76bffc4 100644
--- a/tests/test_batch.py
+++ b/tests/test_batch.py
@@ -21,6 +21,7 @@
 from sqlalchemy import String
 from sqlalchemy import Table
 from sqlalchemy import Text
+from sqlalchemy import TypeDecorator
 from sqlalchemy import UniqueConstraint
 from sqlalchemy.dialects import sqlite as sqlite_dialect
 from sqlalchemy.schema import CreateIndex
@@ -1154,6 +1155,33 @@ def test_change_type(self):
             "ALTER TABLE _alembic_tmp_foo RENAME TO foo",
         )
 
+    def test_change_type_json_typedecorator(self):
+        """a TypeDecorator that augments JSON must not be CAST either.
+
+        CASTing to JSON in SQLite yields 0, so emitting the CAST here
+        would destroy the column's data.
+
+        """
+
+        class CustomJson(TypeDecorator):
+            impl = JSON
+            cache_ok = True
+
+        context = self._fixture()
+        self.table.append_column(Column("toj", Text))
+        with self.op.batch_alter_table(
+            "foo", copy_from=self.table
+        ) as batch_op:
+            batch_op.alter_column("toj", type_=CustomJson)
+        context.assert_(
+            "CREATE TABLE _alembic_tmp_foo (id INTEGER NOT NULL, "
+            "data VARCHAR(50), x INTEGER, toj JSON, PRIMARY KEY (id))",
+            "INSERT INTO _alembic_tmp_foo (id, data, x, toj) "
+            "SELECT foo.id, foo.data, foo.x, foo.toj FROM foo",
+            "DROP TABLE foo",
+            "ALTER TABLE _alembic_tmp_foo RENAME TO foo",
+        )
+
     def test_change_type_from_schematype(self):
         context = self._fixture()
         self.table.append_column(