✨ add support for creating SQLite STRICT tables via -M/--strict CLI switch (#107)

Author: techouse

README.md

@@ -60,6 +60,7 @@ Options:
   -X, --without-foreign-keys      Do not transfer foreign keys.
   -Z, --without-tables            Do not transfer tables, data only.
   -W, --without-data              Do not transfer table data, DDL only.
+  -M, --strict                    Create SQLite STRICT tables when supported.
   -h, --mysql-host TEXT           MySQL host. Defaults to localhost.
   -P, --mysql-port INTEGER        MySQL port. Defaults to 3306.
   --mysql-charset TEXT            MySQL database and table character set

docs/README.rst

@@ -38,6 +38,7 @@ Transfer Options
 - ``-X, --without-foreign-keys``: Do not transfer foreign keys.
 - ``-Z, --without-tables``: Do not transfer tables, data only.
 - ``-W, --without-data``: Do not transfer table data, DDL only.
+- ``-M, --strict``: Create SQLite STRICT tables when supported.
 
 Connection Options
 """"""""""""""""""

src/mysql_to_sqlite3/cli.py

@@ -106,6 +106,12 @@
     is_flag=True,
     help="Do not transfer table data, DDL only.",
 )
+@click.option(
+    "-M",
+    "--strict",
+    is_flag=True,
+    help="Create SQLite STRICT tables when supported.",
+)
 @click.option("-h", "--mysql-host", default="localhost", help="MySQL host. Defaults to localhost.")
 @click.option("-P", "--mysql-port", type=int, default=3306, help="MySQL port. Defaults to 3306.")
 @click.option(
@@ -167,6 +173,7 @@ def cli(
     without_foreign_keys: bool,
     without_tables: bool,
     without_data: bool,
+    strict: bool,
     mysql_host: str,
     mysql_port: int,
     mysql_charset: str,
@@ -215,6 +222,7 @@ def cli(
             without_foreign_keys=without_foreign_keys or bool(mysql_tables) or bool(exclude_mysql_tables),
             without_tables=without_tables,
             without_data=without_data,
+            sqlite_strict=strict,
             mysql_host=mysql_host,
             mysql_port=mysql_port,
             mysql_charset=mysql_charset,

src/mysql_to_sqlite3/transporter.py

@@ -120,8 +120,17 @@ def __init__(self, **kwargs: Unpack[MySQLtoSQLiteParams]) -> None:
 
         self._quiet = bool(kwargs.get("quiet", False))
 
+        self._sqlite_strict = bool(kwargs.get("sqlite_strict", False))
+
         self._logger = self._setup_logger(log_file=kwargs.get("log_file") or None, quiet=self._quiet)
 
+        if self._sqlite_strict and sqlite3.sqlite_version < "3.37.0":
+            self._logger.warning(
+                "SQLite version %s does not support STRICT tables. Tables will be created without strict mode.",
+                sqlite3.sqlite_version,
+            )
+            self._sqlite_strict = False
+
         sqlite3.register_adapter(Decimal, adapt_decimal)
         sqlite3.register_converter("DECIMAL", convert_decimal)
         sqlite3.register_adapter(timedelta, adapt_timedelta)
@@ -570,7 +579,10 @@ def _build_create_table_sql(self, table_name: str) -> str:
                         "ON DELETE {on_delete}".format(**foreign_key)  # type: ignore[str-bytes-safe]
                     )
 
-        sql += "\n);"
+        sql += "\n)"
+        if self._sqlite_strict:
+            sql += " STRICT"
+        sql += ";\n"
         sql += indices
 
         return sql

src/mysql_to_sqlite3/types.py

@@ -39,6 +39,7 @@ class MySQLtoSQLiteParams(TypedDict):
     prefix_indices: t.Optional[bool]
     quiet: t.Optional[bool]
     sqlite_file: t.Union[str, "os.PathLike[t.Any]"]
+    sqlite_strict: t.Optional[bool]
     vacuum: t.Optional[bool]
     without_tables: t.Optional[bool]
     without_data: t.Optional[bool]
@@ -74,6 +75,7 @@ class MySQLtoSQLiteAttributes:
     _sqlite: Connection
     _sqlite_cur: Cursor
     _sqlite_file: t.Union[str, "os.PathLike[t.Any]"]
+    _sqlite_strict: bool
     _without_tables: bool
     _sqlite_json1_extension_enabled: bool
     _vacuum: bool

tests/unit/test_transporter.py

@@ -1,8 +1,11 @@
+import builtins
 import sqlite3
 from unittest.mock import MagicMock, patch
 
 import pytest
+from pytest_mock import MockerFixture
 
+from mysql_to_sqlite3.sqlite_utils import CollatingSequences
 from mysql_to_sqlite3.transporter import MySQLtoSQLite
 
 
@@ -152,6 +155,155 @@ def test_transfer_exception_handling(self, mock_sqlite_connect: MagicMock, mock_
             # Verify that foreign keys are re-enabled in the finally block
             mock_sqlite_cursor.execute.assert_called_with("PRAGMA foreign_keys=ON")
 
+    @patch("mysql_to_sqlite3.transporter.sqlite3.connect")
+    @patch("mysql_to_sqlite3.transporter.mysql.connector.connect")
+    def test_sqlite_strict_supported_keeps_flag(
+        self,
+        mock_mysql_connect: MagicMock,
+        mock_sqlite_connect: MagicMock,
+        mocker: MockerFixture,
+    ) -> None:
+        """Ensure STRICT mode remains enabled when SQLite supports it."""
+
+        class FakeMySQLConnection:
+            def __init__(self) -> None:
+                self.database = None
+
+            def is_connected(self) -> bool:
+                return True
+
+            def cursor(self, *args, **kwargs) -> MagicMock:
+                return MagicMock()
+
+        mock_logger = MagicMock()
+        mocker.patch.object(MySQLtoSQLite, "_setup_logger", return_value=mock_logger)
+        mocker.patch("mysql_to_sqlite3.transporter.sqlite3.sqlite_version", "3.38.0")
+        mock_mysql_connect.return_value = FakeMySQLConnection()
+
+        mock_sqlite_cursor = MagicMock()
+        mock_sqlite_connection = MagicMock()
+        mock_sqlite_connection.cursor.return_value = mock_sqlite_cursor
+        mock_sqlite_connect.return_value = mock_sqlite_connection
+
+        from mysql_to_sqlite3 import transporter as transporter_module
+
+        original_isinstance = builtins.isinstance
+
+        def fake_isinstance(obj: object, classinfo: object) -> bool:
+            if classinfo is transporter_module.MySQLConnectionAbstract:
+                return True
+            return original_isinstance(obj, classinfo)
+
+        mocker.patch("mysql_to_sqlite3.transporter.isinstance", side_effect=fake_isinstance)
+
+        instance = MySQLtoSQLite(
+            sqlite_file="file.db",
+            mysql_user="user",
+            mysql_password=None,
+            mysql_database="db",
+            mysql_host="localhost",
+            mysql_port=3306,
+            sqlite_strict=True,
+        )
+
+        assert instance._sqlite_strict is True
+        mock_logger.warning.assert_not_called()
+
+    @patch("mysql_to_sqlite3.transporter.sqlite3.connect")
+    @patch("mysql_to_sqlite3.transporter.mysql.connector.connect")
+    def test_sqlite_strict_unsupported_disables_flag(
+        self,
+        mock_mysql_connect: MagicMock,
+        mock_sqlite_connect: MagicMock,
+        mocker: MockerFixture,
+    ) -> None:
+        """Ensure STRICT mode is disabled with a warning on old SQLite versions."""
+
+        class FakeMySQLConnection:
+            def __init__(self) -> None:
+                self.database = None
+
+            def is_connected(self) -> bool:
+                return True
+
+            def cursor(self, *args, **kwargs) -> MagicMock:
+                return MagicMock()
+
+        mock_logger = MagicMock()
+        mocker.patch.object(MySQLtoSQLite, "_setup_logger", return_value=mock_logger)
+        mocker.patch("mysql_to_sqlite3.transporter.sqlite3.sqlite_version", "3.36.0")
+        mock_mysql_connect.return_value = FakeMySQLConnection()
+
+        mock_sqlite_cursor = MagicMock()
+        mock_sqlite_connection = MagicMock()
+        mock_sqlite_connection.cursor.return_value = mock_sqlite_cursor
+        mock_sqlite_connect.return_value = mock_sqlite_connection
+
+        from mysql_to_sqlite3 import transporter as transporter_module
+
+        original_isinstance = builtins.isinstance
+
+        def fake_isinstance(obj: object, classinfo: object) -> bool:
+            if classinfo is transporter_module.MySQLConnectionAbstract:
+                return True
+            return original_isinstance(obj, classinfo)
+
+        mocker.patch("mysql_to_sqlite3.transporter.isinstance", side_effect=fake_isinstance)
+
+        instance = MySQLtoSQLite(
+            sqlite_file="file.db",
+            mysql_user="user",
+            mysql_password=None,
+            mysql_database="db",
+            mysql_host="localhost",
+            mysql_port=3306,
+            sqlite_strict=True,
+        )
+
+        assert instance._sqlite_strict is False
+        mock_logger.warning.assert_called_once()
+
+    def test_build_create_table_sql_appends_strict(self) -> None:
+        """Ensure STRICT is appended to CREATE TABLE statements when enabled."""
+        with patch.object(MySQLtoSQLite, "__init__", return_value=None):
+            instance = MySQLtoSQLite()
+
+        instance._sqlite_strict = True
+        instance._sqlite_json1_extension_enabled = False
+        instance._mysql_cur_dict = MagicMock()
+        instance._mysql_cur_dict.fetchall.side_effect = [
+            [
+                {
+                    "Field": "id",
+                    "Type": "INTEGER",
+                    "Null": "NO",
+                    "Default": None,
+                    "Key": "PRI",
+                    "Extra": "auto_increment",
+                },
+                {
+                    "Field": "name",
+                    "Type": "TEXT",
+                    "Null": "NO",
+                    "Default": None,
+                    "Key": "",
+                    "Extra": "",
+                },
+            ],
+            [],
+        ]
+        instance._mysql_cur_dict.fetchone.return_value = {"count": 0}
+        instance._mysql_database = "db"
+        instance._collation = CollatingSequences.BINARY
+        instance._prefix_indices = False
+        instance._without_tables = False
+        instance._without_foreign_keys = True
+        instance._logger = MagicMock()
+
+        sql = instance._build_create_table_sql("products")
+
+        assert "STRICT;" in sql
+
     def test_constructor_missing_mysql_database(self) -> None:
         """Test constructor raises ValueError if mysql_database is missing."""
         from mysql_to_sqlite3.transporter import MySQLtoSQLite