✨ add --without-tables option (#78)

Author: techouse

README.md

@@ -31,7 +31,7 @@ mysql2sqlite --help
 ```
 Usage: mysql2sqlite [OPTIONS]
 
-  Transfer MySQL to SQLite using the provided CLI options.
+  mysql2sqlite version 2.1.12 Copyright (c) 2019-2024 Klemen Tusar
 
 Options:
   -f, --sqlite-file PATH          SQLite3 database file  [required]
@@ -44,27 +44,23 @@ Options:
                                   foreign-keys which inhibits the transfer of
                                   foreign keys. Can not be used together with
                                   --exclude-mysql-tables.
-
   -e, --exclude-mysql-tables TUPLE
                                   Transfer all tables except these specific
                                   tables (space separated table names).
                                   Implies --without-foreign-keys which
                                   inhibits the transfer of foreign keys. Can
                                   not be used together with --mysql-tables.
-
   -L, --limit-rows INTEGER        Transfer only a limited number of rows from
                                   each table.
-
   -C, --collation [BINARY|NOCASE|RTRIM]
                                   Create datatypes of TEXT affinity using a
                                   specified collation sequence.  [default:
                                   BINARY]
-
   -K, --prefix-indices            Prefix indices with their corresponding
                                   tables. This ensures that their names remain
                                   unique across the SQLite database.
-
   -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.
   -h, --mysql-host TEXT           MySQL host. Defaults to localhost.
   -P, --mysql-port INTEGER        MySQL port. Defaults to 3306.
@@ -75,13 +71,11 @@ Options:
   -V, --vacuum                    Use the VACUUM command to rebuild the SQLite
                                   database file, repacking it into a minimal
                                   amount of disk space
-
   --use-buffered-cursors          Use MySQLCursorBuffered for reading the
                                   MySQL database. This can be useful in
                                   situations where multiple queries, with
                                   small result sets, need to be combined or
                                   computed with each other.
-
   -q, --quiet                     Quiet. Display only errors.
   --debug                         Debug mode. Will throw exceptions.
   --version                       Show the version and exit.

docs/README.rst

@@ -36,6 +36,7 @@ Transfer Options
 - ``-C, --collation [BINARY|NOCASE|RTRIM]``: Create datatypes of TEXT affinity using a specified collation sequence. The default is BINARY.
 - ``-K, --prefix-indices``: Prefix indices with their corresponding tables. This ensures that their names remain unique across the SQLite database.
 - ``-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.
 
 Connection Options

src/mysql_to_sqlite3/cli.py

@@ -92,6 +92,12 @@
     "This ensures that their names remain unique across the SQLite database.",
 )
 @click.option("-X", "--without-foreign-keys", is_flag=True, help="Do not transfer foreign keys.")
+@click.option(
+    "-Z",
+    "--without-tables",
+    is_flag=True,
+    help="Do not transfer tables, data only.",
+)
 @click.option(
     "-W",
     "--without-data",
@@ -139,6 +145,7 @@ def cli(
     collation: t.Optional[str],
     prefix_indices: bool,
     without_foreign_keys: bool,
+    without_tables: bool,
     without_data: bool,
     mysql_host: str,
     mysql_port: int,
@@ -154,6 +161,12 @@ def cli(
     """Transfer MySQL to SQLite using the provided CLI options."""
     click.echo(_copyright_header)
     try:
+        # check if both mysql_skip_create_table and mysql_skip_transfer_data are True
+        if without_tables and without_data:
+            raise click.ClickException(
+                "Error: Both -Z/--without-tables and -W/--without-data are set. There is nothing to do. Exiting..."
+            )
+
         if mysql_tables and exclude_mysql_tables:
             raise click.UsageError("Illegal usage: --mysql-tables and --exclude-mysql-tables are mutually exclusive!")
 
@@ -168,6 +181,7 @@ def cli(
             collation=collation,
             prefix_indices=prefix_indices,
             without_foreign_keys=without_foreign_keys or (mysql_tables is not None and len(mysql_tables) > 0),
+            without_tables=without_tables,
             without_data=without_data,
             mysql_host=mysql_host,
             mysql_port=mysql_port,

src/mysql_to_sqlite3/transporter.py

@@ -57,18 +57,18 @@ def __init__(self, **kwargs: tx.Unpack[MySQLtoSQLiteParams]) -> None:
 
         self._mysql_password = str(kwargs.get("mysql_password")) or None
 
-        self._mysql_host = kwargs.get("mysql_host") or "localhost"
+        self._mysql_host = kwargs.get("mysql_host", "localhost") or "localhost"
 
-        self._mysql_port = kwargs.get("mysql_port") or 3306
+        self._mysql_port = kwargs.get("mysql_port", 3306) or 3306
 
         self._mysql_tables = kwargs.get("mysql_tables") or tuple()
 
         self._exclude_mysql_tables = kwargs.get("exclude_mysql_tables") or tuple()
 
-        if len(self._mysql_tables) > 0 and len(self._exclude_mysql_tables) > 0:
+        if bool(self._mysql_tables) and bool(self._exclude_mysql_tables):
             raise ValueError("mysql_tables and exclude_mysql_tables are mutually exclusive")
 
-        self._limit_rows = kwargs.get("limit_rows") or 0
+        self._limit_rows = kwargs.get("limit_rows", 0) or 0
 
         if kwargs.get("collation") is not None and str(kwargs.get("collation")).upper() in {
             CollatingSequences.BINARY,
@@ -79,26 +79,30 @@ def __init__(self, **kwargs: tx.Unpack[MySQLtoSQLiteParams]) -> None:
         else:
             self._collation = CollatingSequences.BINARY
 
-        self._prefix_indices = kwargs.get("prefix_indices") or False
+        self._prefix_indices = kwargs.get("prefix_indices", False) or False
 
-        if len(self._mysql_tables) > 0 or len(self._exclude_mysql_tables) > 0:
+        if bool(self._mysql_tables) or bool(self._exclude_mysql_tables):
             self._without_foreign_keys = True
         else:
-            self._without_foreign_keys = kwargs.get("without_foreign_keys") or False
+            self._without_foreign_keys = bool(kwargs.get("without_foreign_keys", False))
+
+        self._without_data = bool(kwargs.get("without_data", False))
+        self._without_tables = bool(kwargs.get("without_tables", False))
 
-        self._without_data = kwargs.get("without_data") or False
+        if self._without_tables and self._without_data:
+            raise ValueError("Unable to continue without transferring data or creating tables!")
 
-        self._mysql_ssl_disabled = kwargs.get("mysql_ssl_disabled") or False
+        self._mysql_ssl_disabled = bool(kwargs.get("mysql_ssl_disabled", False))
 
         self._current_chunk_number = 0
 
         self._chunk_size = kwargs.get("chunk") or None
 
-        self._buffered = kwargs.get("buffered") or False
+        self._buffered = bool(kwargs.get("buffered", False))
 
-        self._vacuum = kwargs.get("vacuum") or False
+        self._vacuum = bool(kwargs.get("vacuum", False))
 
-        self._quiet = kwargs.get("quiet") or False
+        self._quiet = bool(kwargs.get("quiet", False))
 
         self._logger = self._setup_logger(log_file=kwargs.get("log_file") or None, quiet=self._quiet)
 
@@ -113,7 +117,7 @@ def __init__(self, **kwargs: tx.Unpack[MySQLtoSQLiteParams]) -> None:
 
         self._sqlite_cur = self._sqlite.cursor()
 
-        self._json_as_text = kwargs.get("json_as_text") or False
+        self._json_as_text = bool(kwargs.get("json_as_text", False))
 
         self._sqlite_json1_extension_enabled = not self._json_as_text and self._check_sqlite_json1_extension_enabled()
 
@@ -490,7 +494,7 @@ def _build_create_table_sql(self, table_name: str) -> str:
         sql += primary
         sql = sql.rstrip(", ")
 
-        if not self._without_foreign_keys:
+        if not self._without_tables and not self._without_foreign_keys:
             server_version: t.Optional[t.Tuple[int, ...]] = self._mysql.get_server_version()
             self._mysql_cur_dict.execute(
                 """
@@ -662,16 +666,18 @@ def transfer(self) -> None:
                     table_name = table_name.decode()
 
                 self._logger.info(
-                    "%sTransferring table %s",
+                    "%s%sTransferring table %s",
                     "[WITHOUT DATA] " if self._without_data else "",
+                    "[ONLY DATA] " if self._without_tables else "",
                     table_name,
                 )
 
                 # reset the chunk
                 self._current_chunk_number = 0
 
-                # create the table
-                self._create_table(table_name)  # type: ignore[arg-type]
+                if not self._without_tables:
+                    # create the table
+                    self._create_table(table_name)  # type: ignore[arg-type]
 
                 if not self._without_data:
                     # get the size of the data

src/mysql_to_sqlite3/types.py

@@ -31,6 +31,7 @@ class MySQLtoSQLiteParams(tx.TypedDict):
     quiet: t.Optional[bool]
     sqlite_file: t.Union[str, "os.PathLike[t.Any]"]
     vacuum: t.Optional[bool]
+    without_tables: t.Optional[bool]
     without_data: t.Optional[bool]
     without_foreign_keys: t.Optional[bool]
 
@@ -62,6 +63,7 @@ class MySQLtoSQLiteAttributes:
     _sqlite: Connection
     _sqlite_cur: Cursor
     _sqlite_file: t.Union[str, "os.PathLike[t.Any]"]
+    _without_tables: bool
     _sqlite_json1_extension_enabled: bool
     _vacuum: bool
     _without_data: bool

tests/func/mysql_to_sqlite3_test.py

@@ -199,6 +199,29 @@ def cursor(
             assert any("MySQL Database does not exist!" in message for message in caplog.messages)
         assert "Unknown database" in str(excinfo.value)
 
+    @pytest.mark.init
+    def test_without_tables_and_without_data(
+        self,
+        sqlite_database: "os.PathLike[t.Any]",
+        mysql_database: Database,
+        mysql_credentials: MySQLCredentials,
+        caplog: LogCaptureFixture,
+        tmpdir: LocalPath,
+        faker: Faker,
+    ) -> None:
+        with pytest.raises(ValueError) as excinfo:
+            MySQLtoSQLite(  # type: ignore[call-arg]
+                sqlite_file=sqlite_database,
+                mysql_user=mysql_credentials.user,
+                mysql_password=mysql_credentials.password,
+                mysql_database=mysql_credentials.database,
+                mysql_host=mysql_credentials.host,
+                mysql_port=mysql_credentials.port,
+                without_tables=True,
+                without_data=True,
+            )
+        assert "Unable to continue without transferring data or creating tables!" in str(excinfo.value)
+
     @pytest.mark.xfail
     @pytest.mark.init
     @pytest.mark.parametrize(

tests/func/test_cli.py

@@ -219,6 +219,81 @@ def test_invalid_database_port(
             }
         )
 
+    def test_without_data(
+        self,
+        cli_runner: CliRunner,
+        sqlite_database: "os.PathLike[t.Any]",
+        mysql_database: Database,
+        mysql_credentials: MySQLCredentials,
+    ) -> None:
+        result: Result = cli_runner.invoke(
+            mysql2sqlite,
+            [
+                "-f",
+                str(sqlite_database),
+                "-d",
+                mysql_credentials.database,
+                "-u",
+                mysql_credentials.user,
+                "--mysql-password",
+                mysql_credentials.password,
+                "-h",
+                mysql_credentials.host,
+                "-P",
+                str(mysql_credentials.port),
+                "-W",
+            ],
+        )
+        assert result.exit_code == 0
+
+    def test_without_tables(
+        self,
+        cli_runner: CliRunner,
+        sqlite_database: "os.PathLike[t.Any]",
+        mysql_database: Database,
+        mysql_credentials: MySQLCredentials,
+    ) -> None:
+        # First we need to create the tables in the SQLite database
+        result1: Result = cli_runner.invoke(
+            mysql2sqlite,
+            [
+                "-f",
+                str(sqlite_database),
+                "-d",
+                mysql_credentials.database,
+                "-u",
+                mysql_credentials.user,
+                "--mysql-password",
+                mysql_credentials.password,
+                "-h",
+                mysql_credentials.host,
+                "-P",
+                str(mysql_credentials.port),
+                "-W",
+            ],
+        )
+        assert result1.exit_code == 0
+
+        result2: Result = cli_runner.invoke(
+            mysql2sqlite,
+            [
+                "-f",
+                str(sqlite_database),
+                "-d",
+                mysql_credentials.database,
+                "-u",
+                mysql_credentials.user,
+                "--mysql-password",
+                mysql_credentials.password,
+                "-h",
+                mysql_credentials.host,
+                "-P",
+                str(mysql_credentials.port),
+                "-Z",
+            ],
+        )
+        assert result2.exit_code == 0
+
     @pytest.mark.parametrize(
         "chunk, vacuum, use_buffered_cursors, quiet",
         [