✨ MySQL Unix socket connection support (#137)

MySQL Unix socket connection support

---------

Co-authored-by: Kuba Suder <[email protected]>

Author: techouse

README.md

@@ -49,6 +49,7 @@ Options:
   --mysql-password TEXT           MySQL password
   -h, --mysql-host TEXT           MySQL host. Defaults to localhost.
   -P, --mysql-port INTEGER        MySQL port. Defaults to 3306.
+  -k, --mysql-socket PATH         Path to MySQL unix socket file.
   -S, --skip-ssl                  Disable MySQL connection encryption.
   -i, --mysql-insert-method [DEFAULT|IGNORE|UPDATE]
                                   MySQL insert method. DEFAULT will throw

docs/README.rst

@@ -29,6 +29,7 @@ Connection Options
 - ``-h, --mysql-host TEXT``: MySQL host. Defaults to localhost.
 - ``-P, --mysql-port INTEGER``: MySQL port. Defaults to 3306.
 - ``-S, --skip-ssl``: Disable MySQL connection encryption.
+- ``--mysql-socket TEXT``: Path to MySQL unix socket file.
 
 Transfer Options
 """"""""""""""""

src/sqlite3_to_mysql/cli.py

@@ -62,6 +62,13 @@
 @click.option("--mysql-password", default=None, help="MySQL password")
 @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(
+    "-k",
+    "--mysql-socket",
+    type=click.Path(exists=True),
+    default=None,
+    help="Path to MySQL unix socket file.",
+)
 @click.option("-S", "--skip-ssl", is_flag=True, help="Disable MySQL connection encryption.")
 @click.option(
     "-i",
@@ -137,6 +144,7 @@ def cli(
     mysql_database: str,
     mysql_host: str,
     mysql_port: int,
+    mysql_socket: t.Optional[str],
     skip_ssl: bool,
     mysql_insert_method: str,
     mysql_truncate_tables: bool,
@@ -157,6 +165,9 @@ def cli(
     """Transfer SQLite to MySQL using the provided CLI options."""
     click.echo(_copyright_header)
     try:
+        if mysql_port and mysql_socket:
+            raise click.ClickException("Error: Can only specify either -P/--mysql-port or -k/--mysql-socket, not both.")
+
         if mysql_collation:
             charset_collations: t.Tuple[str, ...] = tuple(
                 cs.collation for cs in mysql_supported_character_sets(mysql_charset.lower())
@@ -183,6 +194,7 @@ def cli(
             mysql_database=mysql_database,
             mysql_host=mysql_host,
             mysql_port=mysql_port,
+            mysql_socket=mysql_socket,
             mysql_ssl_disabled=skip_ssl,
             mysql_insert_method=mysql_insert_method,
             mysql_truncate_tables=mysql_truncate_tables,

src/sqlite3_to_mysql/transporter.py

@@ -75,6 +75,15 @@ def __init__(self, **kwargs: tx.Unpack[SQLite3toMySQLParams]):
 
         self._mysql_port = kwargs.get("mysql_port", 3306) or 3306
 
+        if kwargs.get("mysql_socket") is not None:
+            if not os.path.exists(str(kwargs.get("mysql_socket"))):
+                raise FileNotFoundError("MySQL socket does not exist")
+            else:
+                self._mysql_socket = realpath(str(kwargs.get("mysql_socket")))
+                self._mysql_port = None
+        else:
+            self._mysql_socket = None
+
         self._sqlite_tables = kwargs.get("sqlite_tables") or tuple()
 
         self._without_foreign_keys = bool(self._sqlite_tables) or bool(kwargs.get("without_foreign_keys", False))
@@ -144,6 +153,7 @@ def __init__(self, **kwargs: tx.Unpack[SQLite3toMySQLParams]):
                 password=self._mysql_password,
                 host=self._mysql_host,
                 port=self._mysql_port,
+                unix_socket=self._mysql_socket,
                 ssl_disabled=self._mysql_ssl_disabled,
                 use_pure=True,
                 charset=self._mysql_charset,

src/sqlite3_to_mysql/types.py

@@ -20,6 +20,7 @@ class SQLite3toMySQLParams(tx.TypedDict):
     mysql_password: t.Optional[t.Union[str, bool]]
     mysql_host: t.Optional[str]
     mysql_port: t.Optional[int]
+    mysql_socket: t.Optional[t.Union[str, "os.PathLike[t.Any]"]]
     mysql_ssl_disabled: t.Optional[bool]
     chunk: t.Optional[int]
     quiet: t.Optional[bool]
@@ -48,7 +49,8 @@ class SQLite3toMySQLAttributes:
     _mysql_user: str
     _mysql_password: t.Optional[str]
     _mysql_host: str
-    _mysql_port: int
+    _mysql_port: t.Optional[int]
+    _mysql_socket: t.Optional[t.Union[str, "os.PathLike[t.Any]"]]
     _mysql_ssl_disabled: bool
     _chunk_size: t.Optional[int]
     _quiet: bool

tests/unit/types_test.py

@@ -23,6 +23,7 @@ def test_sqlite3_to_mysql_params_typing(self) -> None:
             "mysql_password": "password",
             "mysql_host": "localhost",
             "mysql_port": 3306,
+            "mysql_socket": "/var/run/mysqld/mysqld.sock",
             "mysql_ssl_disabled": True,
             "chunk": 1000,
             "quiet": False,
@@ -50,6 +51,7 @@ def test_sqlite3_to_mysql_params_typing(self) -> None:
         assert params["mysql_password"] == "password"
         assert params["mysql_host"] == "localhost"
         assert params["mysql_port"] == 3306
+        assert params["mysql_socket"] == "/var/run/mysqld/mysqld.sock"
         assert params["mysql_ssl_disabled"] is True
         assert params["chunk"] == 1000
         assert params["quiet"] is False
@@ -90,6 +92,7 @@ def __init__(self) -> None:
                 self._mysql_password = "password"
                 self._mysql_host = "localhost"
                 self._mysql_port = 3306
+                self._mysql_socket = "/var/run/mysqld/mysqld.sock"
                 self._mysql_ssl_disabled = True
                 self._chunk_size = 1000
                 self._quiet = False
@@ -129,6 +132,7 @@ def __init__(self) -> None:
         assert instance._mysql_password == "password"
         assert instance._mysql_host == "localhost"
         assert instance._mysql_port == 3306
+        assert instance._mysql_socket == "/var/run/mysqld/mysqld.sock"
         assert instance._mysql_ssl_disabled is True
         assert instance._chunk_size == 1000
         assert instance._quiet is False