Fix: SQLite STRICT is not propagated in batch mode (#1758)

alembic/ddl/impl.py

@@ -688,6 +688,14 @@ def autogen_column_reflect(self, inspector, table, column_info):
 
         """
 
+    def autogen_table_reflect(self, inspector, table):
+        """A hook that is called when a Table is reflected from the
+        database during the autogenerate process.
+
+        Dialects can elect to modify the information gathered here.
+
+        """
+
     def start_migrations(self) -> None:
         """A hook called when :meth:`.EnvironmentContext.run_migrations`
         is called.

alembic/ddl/sqlite.py

@@ -160,6 +160,17 @@ def autogen_column_reflect(
         ):
             column_info["default"] = "(%s)" % (column_info["default"],)
 
+    def autogen_table_reflect(self, inspector, table):
+        sql_text = sql.text(
+            "SELECT sql FROM sqlite_master WHERE name=:name AND type='table'"
+        )
+        res = inspector.bind.execute(sql_text, {"name": table.name}).scalar()
+        if res:
+            if re.search(r"\bSTRICT\b\s*;?\s*$", res, re.I):
+                table.kwargs["sqlite_strict"] = True
+            if re.search(r"\bWITHOUT ROWID\b", res, re.I):
+                table.kwargs["sqlite_with_rowid"] = False
+
     def render_ddl_sql_expr(
         self, expr: ClauseElement, is_server_default: bool = False, **kw
     ) -> str:

alembic/operations/batch.py

@@ -146,6 +146,12 @@ def flush(self) -> None:
                         *self.reflect_args,
                         **self.reflect_kwargs,
                     )
+                    from sqlalchemy import inspect as sqla_inspect
+
+                    self.operations.impl.autogen_table_reflect(
+                        sqla_inspect(self.operations.get_bind()),
+                        existing_table,
+                    )
                     reflected = True
 
                 batch_impl = ApplyBatchImpl(
@@ -642,7 +648,8 @@ def drop_column(
             and kw["existing_type"].name  # type:ignore[attr-defined]
         ):
             self.named_constraints.pop(
-                kw["existing_type"].name, None  # type:ignore[attr-defined]
+                kw["existing_type"].name,
+                None,  # type:ignore[attr-defined]
             )
 
     def create_column_comment(self, column):

tests/test_batch.py

@@ -346,8 +346,8 @@ def _assert_impl(
                 "INSERT INTO %(schema)s%(temp_name)s (%(colnames)s) "
                 "SELECT %(tname_colnames)s FROM %(schema)stname" % args,
                 "DROP TABLE %(schema)stname" % args,
-                "ALTER TABLE %(schema)s%(temp_name)s "
-                "RENAME TO %(schema)stname" % args,
+                "ALTER TABLE %(schema)s%(temp_name)s RENAME TO %(schema)stname"
+                % args,
             ]
         )
         if idx_stmt:
@@ -775,8 +775,9 @@ def test_add_fk(self):
         new_table = self._assert_impl(
             impl,
             colnames=["id", "x", "y", "user_id"],
-            ddl_contains="CONSTRAINT fk1 FOREIGN KEY(user_id) "
-            'REFERENCES "user" (id)',
+            ddl_contains=(
+                "CONSTRAINT fk1 FOREIGN KEY(user_id) " 'REFERENCES "user" (id)'
+            ),
         )
         eq_(
             list(new_table.c.user_id.foreign_keys)[0]._get_colspec(), "user.id"
@@ -1589,8 +1590,7 @@ def test_fk_points_to_me_recreate(self):
 
     @exclusions.only_on("sqlite")
     @exclusions.fails(
-        "intentionally asserting that this "
-        "doesn't work w/ pragma foreign keys"
+        "intentionally asserting that this doesn't work w/ pragma foreign keys"
     )
     def test_fk_points_to_me_sqlite_refinteg(self):
         with self._sqlite_referential_integrity():
@@ -1635,8 +1635,7 @@ def test_selfref_fk_recreate(self):
 
     @exclusions.only_on("sqlite")
     @exclusions.fails(
-        "intentionally asserting that this "
-        "doesn't work w/ pragma foreign keys"
+        "intentionally asserting that this doesn't work w/ pragma foreign keys"
     )
     def test_selfref_fk_sqlite_refinteg(self):
         with self._sqlite_referential_integrity():
@@ -2298,6 +2297,45 @@ def test_create_drop_index(self):
         insp = inspect(self.conn)
         eq_(insp.get_indexes("foo"), [])
 
+    def test_sqlite_batch_strict(self):
+        """test that STRICT is persisted in batch mode.  See #1758"""
+        t = Table(
+            "t",
+            self.metadata,
+            Column("id", Integer, primary_key=True),
+            Column("data", Integer),
+            sqlite_strict=True,
+        )
+        with self.conn.begin():
+            t.create(self.conn)
+
+        with self.op.batch_alter_table("t", recreate="always") as batch_op:
+            batch_op.drop_column("data")
+
+        sql = self.conn.scalar(
+            text("SELECT sql FROM sqlite_master WHERE name='t'")
+        )
+        assert "STRICT" in sql
+
+    def test_sqlite_batch_without_rowid(self):
+        """test that WITHOUT ROWID is persisted in batch mode.  See #1758"""
+        t2 = Table(
+            "t2",
+            self.metadata,
+            Column("id", Integer, primary_key=True),
+            sqlite_with_rowid=False,
+        )
+        with self.conn.begin():
+            t2.create(self.conn)
+
+        with self.op.batch_alter_table("t2", recreate="always") as batch_op:
+            batch_op.add_column(Column("new_col", Integer))
+
+        sql = self.conn.scalar(
+            text("SELECT sql FROM sqlite_master WHERE name='t2'")
+        )
+        assert "WITHOUT ROWID" in sql
+
 
 class BatchRoundTripMySQLTest(BatchRoundTripTest):
     __only_on__ = "mysql", "mariadb"