From d6bae0858deede9ed5886137ea81542243f8d55c Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Mon, 15 Sep 2025 20:06:52 +0100 Subject: [PATCH 01/17] :sparkles: translate SQLite DEFAULT expressions to MySQL-compatible format --- src/sqlite3_to_mysql/transporter.py | 118 +++++++++++++++++++++++++++- 1 file changed, 117 insertions(+), 1 deletion(-) diff --git a/src/sqlite3_to_mysql/transporter.py b/src/sqlite3_to_mysql/transporter.py index 964c8d6..7e419dd 100644 --- a/src/sqlite3_to_mysql/transporter.py +++ b/src/sqlite3_to_mysql/transporter.py @@ -332,6 +332,122 @@ def _translate_type_from_sqlite_to_mysql(self, column_type: str) -> str: return self._mysql_string_type return full_column_type + @staticmethod + def _strip_wrapping_parentheses(expr: str) -> str: + """Remove one or more layers of wrapping parentheses around an expression.""" + s = expr.strip() + while s.startswith("(") and s.endswith(")"): + depth = 0 + balanced = True + for ch in s: + if ch == "(": + depth += 1 + elif ch == ")": + depth -= 1 + if depth < 0: + balanced = False + break + if balanced and depth == 0: + s = s[1:-1].strip() + else: + break + return s + + def _translate_default_for_mysql(self, column_type: str, default: str) -> str: + """Translate SQLite DEFAULT expression to a MySQL-compatible one for common cases. + + Returns a string suitable to append after "DEFAULT ", without the word itself. + Keeps literals as-is, maps `CURRENT_*`/`datetime('now')`/`strftime(...,'now')` to + the appropriate MySQL `CURRENT_*` functions, preserves fractional seconds if the + column type declares a precision, and normalizes booleans to 0/1. + """ + raw = default.strip() + if not raw: + return raw + + s = self._strip_wrapping_parentheses(raw) + u = s.upper() + + # NULL passthrough + if u == "NULL": + return "NULL" + + # Determine base data type + match = self._valid_column_type(column_type) + base = match.group(0).upper() if match else column_type.upper() + + # Regex helpers + current_ts = re.compile(r"^CURRENT_TIMESTAMP(?:\s*\(\s*\))?$", re.IGNORECASE) + current_date = re.compile(r"^CURRENT_DATE(?:\s*\(\s*\))?$", re.IGNORECASE) + current_time = re.compile(r"^CURRENT_TIME(?:\s*\(\s*\))?$", re.IGNORECASE) + sqlite_now_func = re.compile( + r"^(datetime|date|time)\s*\(\s*'now'(?:\s*,\s*'(localtime|utc)')?\s*\)$", + re.IGNORECASE, + ) + strftime_now = re.compile( + r"^strftime\s*\(\s*'([^']+)'\s*,\s*'now'(?:\s*,\s*'(localtime|utc)')?\s*\)$", + re.IGNORECASE, + ) + + # TIMESTAMP/DATETIME + if base.startswith("TIMESTAMP") or base.startswith("DATETIME"): + if current_ts.match(s) or sqlite_now_func.match(s) or strftime_now.match(s): + len_match = self.COLUMN_LENGTH_PATTERN.search(column_type) + fsp = "" + if len_match: + try: + n = int(len_match.group(0).strip("()")) + if 0 < n <= 6: + fsp = f"({n})" + except Exception: + pass + return f"CURRENT_TIMESTAMP{fsp}" + + # DATE + if base.startswith("DATE"): + if ( + current_date.match(s) + or (sqlite_now_func.match(s) and s.lower().startswith("date")) + or strftime_now.match(s) + ): + return "CURRENT_DATE" + + # TIME + if base.startswith("TIME"): + if ( + current_time.match(s) + or (sqlite_now_func.match(s) and s.lower().startswith("time")) + or strftime_now.match(s) + ): + len_match = self.COLUMN_LENGTH_PATTERN.search(column_type) + fsp = "" + if len_match: + try: + n = int(len_match.group(0).strip("()")) + if 0 < n <= 6: + fsp = f"({n})" + except Exception: + pass + return f"CURRENT_TIME{fsp}" + + # Booleans (store as 0/1) + if base in {"BOOL", "BOOLEAN"} or base.startswith("TINYINT"): + if u in {"TRUE", "'TRUE'", '"TRUE"'}: + return "1" + if u in {"FALSE", "'FALSE'", '"FALSE"'}: + return "0" + + # Numeric literals (possibly wrapped) + if re.match(r"^[+-]?\d+(\.\d+)?$", s): + return s + + # Quoted strings and hex blobs pass through as-is + if (s.startswith("'") and s.endswith("'")) or (s.startswith('"') and s.endswith('"')) or u.startswith("X'"): + return s + + # Fallback: return stripped expression (MySQL 8.0.13+ allows expression defaults) + return s + @classmethod def _column_type_length(cls, column_type: str, default: t.Optional[t.Union[str, int, float]] = None) -> str: suffix: t.Optional[t.Match[str]] = cls.COLUMN_LENGTH_PATTERN.search(column_type) @@ -385,7 +501,7 @@ def _create_table(self, table_name: str, transfer_rowid: bool = False) -> None: notnull="NOT NULL" if column["notnull"] or column["pk"] else "NULL", auto_increment="AUTO_INCREMENT" if auto_increment else "", default=( - "DEFAULT " + column["dflt_value"] + "DEFAULT " + self._translate_default_for_mysql(column_type, str(column["dflt_value"])) if column["dflt_value"] and column_type not in MYSQL_COLUMN_TYPES_WITHOUT_DEFAULT and not auto_increment From b0bf5a67c36a394d2caaae644e7ab25f4e3e1993 Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Mon, 15 Sep 2025 22:16:03 +0100 Subject: [PATCH 02/17] :safety_vest: handle ValueError when parsing length for MySQL timestamp and time formats --- src/sqlite3_to_mysql/transporter.py | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/src/sqlite3_to_mysql/transporter.py b/src/sqlite3_to_mysql/transporter.py index 7e419dd..2fed006 100644 --- a/src/sqlite3_to_mysql/transporter.py +++ b/src/sqlite3_to_mysql/transporter.py @@ -397,10 +397,10 @@ def _translate_default_for_mysql(self, column_type: str, default: str) -> str: if len_match: try: n = int(len_match.group(0).strip("()")) - if 0 < n <= 6: - fsp = f"({n})" - except Exception: - pass + except ValueError: + n = None + if n is not None and 0 < n <= 6: + fsp = f"({n})" return f"CURRENT_TIMESTAMP{fsp}" # DATE @@ -424,10 +424,10 @@ def _translate_default_for_mysql(self, column_type: str, default: str) -> str: if len_match: try: n = int(len_match.group(0).strip("()")) - if 0 < n <= 6: - fsp = f"({n})" - except Exception: - pass + except ValueError: + n = None + if n is not None and 0 < n <= 6: + fsp = f"({n})" return f"CURRENT_TIME{fsp}" # Booleans (store as 0/1) From b181eaca6016b9c6268aa39dc4fa98d56dff1b90 Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Thu, 18 Sep 2025 20:34:59 +0100 Subject: [PATCH 03/17] :sparkles: improve _strip_wrapping_parentheses to only remove fully wrapping parentheses --- src/sqlite3_to_mysql/transporter.py | 34 +++++++++++++++++++---------- 1 file changed, 22 insertions(+), 12 deletions(-) diff --git a/src/sqlite3_to_mysql/transporter.py b/src/sqlite3_to_mysql/transporter.py index 2358cf4..cf2558b 100644 --- a/src/sqlite3_to_mysql/transporter.py +++ b/src/sqlite3_to_mysql/transporter.py @@ -341,23 +341,33 @@ def _translate_type_from_sqlite_to_mysql(self, column_type: str) -> str: @staticmethod def _strip_wrapping_parentheses(expr: str) -> str: - """Remove one or more layers of wrapping parentheses around an expression.""" - s = expr.strip() - while s.startswith("(") and s.endswith(")"): - depth = 0 - balanced = True - for ch in s: + """Remove one or more layers of *fully wrapping* parentheses around an expression. + + Only strip if the matching ')' for the very first '(' is the final character + of the string. This avoids corrupting expressions like "(a) + (b)". + """ + s: str = expr.strip() + while s.startswith("("): + depth: int = 0 + match_idx: int = -1 + i: int + ch: str + # Find the matching ')' for the '(' at index 0 + for i, ch in enumerate(s): if ch == "(": depth += 1 elif ch == ")": depth -= 1 - if depth < 0: - balanced = False + if depth == 0: + match_idx = i break - if balanced and depth == 0: - s = s[1:-1].strip() - else: - break + # Only strip if the match closes at the very end + if match_idx == len(s) - 1: + s = s[1:match_idx].strip() + # continue to try stripping more fully-wrapping layers + continue + # Not a fully-wrapped expression; stop + break return s def _translate_default_for_mysql(self, column_type: str, default: str) -> str: From 3ab92d2f4832d79087f34f8e84cc5cf0226c716e Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Thu, 18 Sep 2025 20:35:13 +0100 Subject: [PATCH 04/17] :white_check_mark: add tests for _strip_wrapping_parentheses to verify behavior with various expressions --- tests/unit/sqlite3_to_mysql_test.py | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/tests/unit/sqlite3_to_mysql_test.py b/tests/unit/sqlite3_to_mysql_test.py index 2a35de4..7f4bc30 100644 --- a/tests/unit/sqlite3_to_mysql_test.py +++ b/tests/unit/sqlite3_to_mysql_test.py @@ -633,3 +633,24 @@ def execute(self, statement): # Verify both FOREIGN_KEY_CHECKS statements were executed assert "FOREIGN_KEY_CHECKS=0" in fake_cursor.execute_calls[0] assert "FOREIGN_KEY_CHECKS=1" in fake_cursor.execute_calls[-1] + + @pytest.mark.parametrize( + "expr, expected", + [ + ("a", "a"), + ("(a)", "a"), + ("((a))", "a"), + ("(((a)))", "a"), + ("(a) + (b)", "(a) + (b)"), # not fully wrapped; must remain unchanged + ("((a) + (b))", "(a) + (b)"), # fully wrapped once; strip one layer only + (" ( ( a + b ) ) ", "a + b"), # trims whitespace between iterations + ("((CURRENT_TIMESTAMP))", "CURRENT_TIMESTAMP"), # multiple full layers + ("", ""), # empty remains empty + (" ", ""), # whitespace-only becomes empty + ("(a", "(a"), # unmatched; unchanged + ("a)", "a)"), # unmatched; unchanged + ], + ) + def test_strip_wrapping_parentheses(self, expr: str, expected: str) -> None: + """Verify only fully wrapping outer parentheses are removed, repeatedly.""" + assert SQLite3toMySQL._strip_wrapping_parentheses(expr) == expected From b8deb8968a9ca15afe7de2fd24cbda370f27b04c Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Thu, 18 Sep 2025 21:26:48 +0100 Subject: [PATCH 05/17] :sparkles: add MySQL version checks for expression defaults, CURRENT_TIMESTAMP, and fractional seconds support --- src/sqlite3_to_mysql/mysql_utils.py | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/src/sqlite3_to_mysql/mysql_utils.py b/src/sqlite3_to_mysql/mysql_utils.py index 08d8d45..fcafe64 100644 --- a/src/sqlite3_to_mysql/mysql_utils.py +++ b/src/sqlite3_to_mysql/mysql_utils.py @@ -139,6 +139,30 @@ def check_mysql_fulltext_support(version_string: str) -> bool: return mysql_version >= version.parse("5.6.0") +def check_mysql_expression_defaults_support(version_string: str) -> bool: + """Check for expression defaults support.""" + mysql_version: version.Version = get_mysql_version(version_string) + if "-mariadb" in version_string.lower(): + return mysql_version >= version.parse("10.2.0") + return mysql_version >= version.parse("8.0.13") + + +def check_mysql_current_timestamp_datetime_support(version_string: str) -> bool: + """Check for CURRENT_TIMESTAMP support for DATETIME fields.""" + mysql_version: version.Version = get_mysql_version(version_string) + if "-mariadb" in version_string.lower(): + return mysql_version >= version.parse("10.0.1") + return mysql_version >= version.parse("5.6.5") + + +def check_mysql_fractional_seconds_support(version_string: str) -> bool: + """Check for fractional seconds support.""" + mysql_version: version.Version = get_mysql_version(version_string) + if "-mariadb" in version_string.lower(): + return mysql_version >= version.parse("10.1.2") + return mysql_version >= version.parse("5.6.4") + + def safe_identifier_length(identifier_name: str, max_length: int = 64) -> str: """https://dev.mysql.com/doc/refman/8.0/en/identifier-length.html.""" return str(identifier_name)[:max_length] From 03e3cdf966ef556fca9d783c5c30b52794a682e7 Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Thu, 18 Sep 2025 21:26:57 +0100 Subject: [PATCH 06/17] :sparkles: add flags for expression defaults, CURRENT_TIMESTAMP, and fractional seconds support --- src/sqlite3_to_mysql/types.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/sqlite3_to_mysql/types.py b/src/sqlite3_to_mysql/types.py index 3a5c903..3717877 100644 --- a/src/sqlite3_to_mysql/types.py +++ b/src/sqlite3_to_mysql/types.py @@ -85,3 +85,6 @@ class SQLite3toMySQLAttributes: _mysql_version: str _mysql_json_support: bool _mysql_fulltext_support: bool + _allow_expr_defaults: bool + _allow_current_ts_dt: bool + _allow_fsp: bool From 277dc15f9b8464d7d86a2f40a2a03fc59012a327 Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Thu, 18 Sep 2025 21:27:01 +0100 Subject: [PATCH 07/17] :sparkles: add support for MySQL CURRENT_TIMESTAMP, CURRENT_DATE, and CURRENT_TIME in default expressions --- src/sqlite3_to_mysql/transporter.py | 128 ++++++++++++++++------------ 1 file changed, 74 insertions(+), 54 deletions(-) diff --git a/src/sqlite3_to_mysql/transporter.py b/src/sqlite3_to_mysql/transporter.py index cf2558b..0e16438 100644 --- a/src/sqlite3_to_mysql/transporter.py +++ b/src/sqlite3_to_mysql/transporter.py @@ -44,6 +44,9 @@ MYSQL_INSERT_METHOD, MYSQL_TEXT_COLUMN_TYPES, MYSQL_TEXT_COLUMN_TYPES_WITH_JSON, + check_mysql_current_timestamp_datetime_support, + check_mysql_expression_defaults_support, + check_mysql_fractional_seconds_support, check_mysql_fulltext_support, check_mysql_json_support, check_mysql_values_alias_support, @@ -59,6 +62,18 @@ class SQLite3toMySQL(SQLite3toMySQLAttributes): COLUMN_LENGTH_PATTERN: t.Pattern[str] = re.compile(r"\(\d+\)") COLUMN_PRECISION_AND_SCALE_PATTERN: t.Pattern[str] = re.compile(r"\(\d+,\d+\)") COLUMN_UNSIGNED_PATTERN: t.Pattern[str] = re.compile(r"\bUNSIGNED\b", re.IGNORECASE) + CURRENT_TS: t.Pattern[str] = re.compile(r"^CURRENT_TIMESTAMP(?:\s*\(\s*\))?$", re.IGNORECASE) + CURRENT_DATE: t.Pattern[str] = re.compile(r"^CURRENT_DATE(?:\s*\(\s*\))?$", re.IGNORECASE) + CURRENT_TIME: t.Pattern[str] = re.compile(r"^CURRENT_TIME(?:\s*\(\s*\))?$", re.IGNORECASE) + SQLITE_NOW_FUNC: t.Pattern[str] = re.compile( + r"^(datetime|date|time)\s*\(\s*'now'(?:\s*,\s*'(localtime|utc)')?\s*\)$", + re.IGNORECASE, + ) + STRFTIME_NOW: t.Pattern[str] = re.compile( + r"^strftime\s*\(\s*'([^']+)'\s*,\s*'now'(?:\s*,\s*'(localtime|utc)')?\s*\)$", + re.IGNORECASE, + ) + NUMERIC_LITERAL_PATTERN: t.Pattern[str] = re.compile(r"^[+-]?\d+(?:\.\d+)?(?:[eE][+-]?\d+)?$") MYSQL_CONNECTOR_VERSION: version.Version = version.parse(mysql_connector_version_string) @@ -194,6 +209,9 @@ def __init__(self, **kwargs: Unpack[SQLite3toMySQLParams]): self._mysql_version = self._get_mysql_version() self._mysql_json_support = check_mysql_json_support(self._mysql_version) self._mysql_fulltext_support = check_mysql_fulltext_support(self._mysql_version) + self._allow_expr_defaults = check_mysql_expression_defaults_support(self._mysql_version) + self._allow_current_ts_dt = check_mysql_current_timestamp_datetime_support(self._mysql_version) + self._allow_fsp = check_mysql_fractional_seconds_support(self._mysql_version) if self._use_fulltext and not self._mysql_fulltext_support: raise ValueError("Your MySQL version does not support InnoDB FULLTEXT indexes!") @@ -378,74 +396,76 @@ def _translate_default_for_mysql(self, column_type: str, default: str) -> str: the appropriate MySQL `CURRENT_*` functions, preserves fractional seconds if the column type declares a precision, and normalizes booleans to 0/1. """ - raw = default.strip() + raw: str = default.strip() if not raw: return raw - s = self._strip_wrapping_parentheses(raw) - u = s.upper() + s: str = self._strip_wrapping_parentheses(raw) + u: str = s.upper() # NULL passthrough if u == "NULL": return "NULL" # Determine base data type - match = self._valid_column_type(column_type) - base = match.group(0).upper() if match else column_type.upper() - - # Regex helpers - current_ts = re.compile(r"^CURRENT_TIMESTAMP(?:\s*\(\s*\))?$", re.IGNORECASE) - current_date = re.compile(r"^CURRENT_DATE(?:\s*\(\s*\))?$", re.IGNORECASE) - current_time = re.compile(r"^CURRENT_TIME(?:\s*\(\s*\))?$", re.IGNORECASE) - sqlite_now_func = re.compile( - r"^(datetime|date|time)\s*\(\s*'now'(?:\s*,\s*'(localtime|utc)')?\s*\)$", - re.IGNORECASE, - ) - strftime_now = re.compile( - r"^strftime\s*\(\s*'([^']+)'\s*,\s*'now'(?:\s*,\s*'(localtime|utc)')?\s*\)$", - re.IGNORECASE, - ) + match: t.Optional[re.Match[str]] = self._valid_column_type(column_type) + base: str = match.group(0).upper() if match else column_type.upper() # TIMESTAMP/DATETIME - if base.startswith("TIMESTAMP") or base.startswith("DATETIME"): - if current_ts.match(s) or sqlite_now_func.match(s) or strftime_now.match(s): - len_match = self.COLUMN_LENGTH_PATTERN.search(column_type) - fsp = "" - if len_match: - try: - n = int(len_match.group(0).strip("()")) - except ValueError: - n = None - if n is not None and 0 < n <= 6: - fsp = f"({n})" - return f"CURRENT_TIMESTAMP{fsp}" + if ( + (base.startswith("TIMESTAMP") or base.startswith("DATETIME")) + and (self.CURRENT_TS.match(s) or self.SQLITE_NOW_FUNC.match(s) or self.STRFTIME_NOW.match(s)) + and self._allow_current_ts_dt + ): + # Too old for CURRENT_TIMESTAMP defaults on DATETIME/TIMESTAMP → fall back + len_match: t.Optional[re.Match[str]] = self.COLUMN_LENGTH_PATTERN.search(column_type) + fsp: str = "" + n: t.Optional[int] + if self._allow_fsp and len_match: + try: + n = int(len_match.group(0).strip("()")) + except ValueError: + n = None + if n is not None and 0 < n <= 6: + fsp = f"({n})" + return f"CURRENT_TIMESTAMP{fsp}" # DATE - if base.startswith("DATE"): - if ( - current_date.match(s) - or (sqlite_now_func.match(s) and s.lower().startswith("date")) - or strftime_now.match(s) - ): - return "CURRENT_DATE" + if ( + base.startswith("DATE") + and ( + self.CURRENT_DATE.match(s) + or self.CURRENT_TS.match(s) # map CURRENT_TIMESTAMP → CURRENT_DATE for DATE + or (self.SQLITE_NOW_FUNC.match(s) and s.lower().startswith("date")) + or self.STRFTIME_NOW.match(s) + ) + and self._allow_expr_defaults + ): + # Too old for expression defaults on DATE → fall back + return "CURRENT_DATE" # TIME - if base.startswith("TIME"): - if ( - current_time.match(s) - or (sqlite_now_func.match(s) and s.lower().startswith("time")) - or strftime_now.match(s) - ): - len_match = self.COLUMN_LENGTH_PATTERN.search(column_type) - fsp = "" - if len_match: - try: - n = int(len_match.group(0).strip("()")) - except ValueError: - n = None - if n is not None and 0 < n <= 6: - fsp = f"({n})" - return f"CURRENT_TIME{fsp}" + if ( + base.startswith("TIME") + and ( + self.CURRENT_TIME.match(s) + or self.CURRENT_TS.match(s) # map CURRENT_TIMESTAMP → CURRENT_TIME for TIME + or (self.SQLITE_NOW_FUNC.match(s) and s.lower().startswith("time")) + or self.STRFTIME_NOW.match(s) + ) + and self._allow_expr_defaults + ): + # Too old for expression defaults on TIME → fall back + len_match = self.COLUMN_LENGTH_PATTERN.search(column_type) + fsp = "" + if self._allow_fsp and len_match: + try: + n = int(len_match.group(0).strip("()")) + except ValueError: + n = None + if n is not None and 0 < n <= 6: + fsp = f"({n})" + return f"CURRENT_TIME{fsp}" # Booleans (store as 0/1) if base in {"BOOL", "BOOLEAN"} or base.startswith("TINYINT"): @@ -455,7 +475,7 @@ def _translate_default_for_mysql(self, column_type: str, default: str) -> str: return "0" # Numeric literals (possibly wrapped) - if re.match(r"^[+-]?\d+(\.\d+)?$", s): + if self.NUMERIC_LITERAL_PATTERN.match(s): return s # Quoted strings and hex blobs pass through as-is From 0e9895e8e7eae19250b8010b7d5130d071837c0a Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Thu, 18 Sep 2025 21:27:14 +0100 Subject: [PATCH 08/17] :white_check_mark: add tests for translating SQLite default values to MySQL, covering various data types and expressions --- tests/unit/sqlite3_to_mysql_test.py | 63 +++++++++++++++++++++++++++++ 1 file changed, 63 insertions(+) diff --git a/tests/unit/sqlite3_to_mysql_test.py b/tests/unit/sqlite3_to_mysql_test.py index 7f4bc30..7191599 100644 --- a/tests/unit/sqlite3_to_mysql_test.py +++ b/tests/unit/sqlite3_to_mysql_test.py @@ -654,3 +654,66 @@ def execute(self, statement): def test_strip_wrapping_parentheses(self, expr: str, expected: str) -> None: """Verify only fully wrapping outer parentheses are removed, repeatedly.""" assert SQLite3toMySQL._strip_wrapping_parentheses(expr) == expected + + @staticmethod + def _mk(*, expr: bool, ts_dt: bool, fsp: bool) -> SQLite3toMySQL: + """ + Build a lightweight instance without hitting __init__ (no DB connection needed). + Toggle the same feature flags transporter sets after version checks. + """ + o = SQLite3toMySQL.__new__(SQLite3toMySQL) + o._allow_expr_defaults = expr # MySQL >= 8.0.13 + o._allow_current_ts_dt = ts_dt # MySQL >= 5.6.5 + o._allow_fsp = fsp # MySQL >= 5.6.4 + return o + + @pytest.mark.parametrize( + "col, default, flags, expected", + [ + # --- TIMESTAMP/DATETIME + CURRENT_TIMESTAMP / now() mapping --- + # Too old for CURRENT_TIMESTAMP on TIMESTAMP: fall back to stripped expr + ("TIMESTAMP(3)", "CURRENT_TIMESTAMP", dict(expr=False, ts_dt=False, fsp=False), "CURRENT_TIMESTAMP"), + # Allowed, but no FSP support + ("TIMESTAMP(3)", "CURRENT_TIMESTAMP", dict(expr=False, ts_dt=True, fsp=False), "CURRENT_TIMESTAMP"), + # Allowed with FSP support -> keep precision + ("TIMESTAMP(3)", "CURRENT_TIMESTAMP", dict(expr=False, ts_dt=True, fsp=True), "CURRENT_TIMESTAMP(3)"), + # SQLite-style now -> map to CURRENT_TIMESTAMP (with FSP when allowed) + ("DATETIME(2)", "datetime('now')", dict(expr=False, ts_dt=True, fsp=True), "CURRENT_TIMESTAMP(2)"), + # --- DATE mapping (from 'now' forms or CURRENT_TIMESTAMP) --- + # Only map when expression defaults are allowed + ("DATE", "datetime('now')", dict(expr=True, ts_dt=False, fsp=False), "CURRENT_DATE"), + ("DATE", "datetime('now')", dict(expr=False, ts_dt=False, fsp=False), "datetime('now')"), + ("DATE", "CURRENT_TIMESTAMP", dict(expr=True, ts_dt=True, fsp=True), "CURRENT_DATE"), + ("DATE", "CURRENT_TIMESTAMP", dict(expr=False, ts_dt=True, fsp=True), "CURRENT_TIMESTAMP"), + # --- TIME mapping (from 'now' forms or CURRENT_TIMESTAMP) --- + ("TIME(3)", "CURRENT_TIME", dict(expr=True, ts_dt=False, fsp=True), "CURRENT_TIME(3)"), + ("TIME(3)", "CURRENT_TIME", dict(expr=True, ts_dt=False, fsp=False), "CURRENT_TIME"), + ("TIME(6)", "CURRENT_TIMESTAMP", dict(expr=True, ts_dt=True, fsp=True), "CURRENT_TIME(6)"), + ("TIME(6)", "CURRENT_TIMESTAMP", dict(expr=False, ts_dt=True, fsp=True), "CURRENT_TIMESTAMP"), + # --- Boolean normalization (for BOOL/BOOLEAN/TINYINT) --- + ("BOOLEAN", "TRUE", dict(expr=False, ts_dt=False, fsp=False), "1"), + ("TINYINT(1)", "'FALSE'", dict(expr=False, ts_dt=False, fsp=False), "0"), + # --- Numeric literals (incl. scientific notation) --- + ("INT", "42", dict(expr=False, ts_dt=False, fsp=False), "42"), + ("DOUBLE", "-3.14", dict(expr=False, ts_dt=False, fsp=False), "-3.14"), + ("DOUBLE", "1e-3", dict(expr=False, ts_dt=False, fsp=False), "1e-3"), + ("DOUBLE", "-2.5E+10", dict(expr=False, ts_dt=False, fsp=False), "-2.5E+10"), + # --- Quoted strings and hex blobs pass through unchanged --- + ("VARCHAR(10)", "'hello'", dict(expr=False, ts_dt=False, fsp=False), "'hello'"), + ("BLOB", "X'ABCD'", dict(expr=False, ts_dt=False, fsp=False), "X'ABCD'"), + # --- Expression fallback (strip fully wrapping parens, leave the expr) --- + ("VARCHAR(10)", "(1+2)", dict(expr=False, ts_dt=False, fsp=False), "1+2"), + ], + ) + def test_translate_default_for_mysql(self, col, default, flags, expected): + assert self._mk(**flags)._translate_default_for_mysql(col, default) == expected + + def test_time_mapping_from_sqlite_now_respects_fsp(self): + assert ( + self._mk(expr=True, ts_dt=False, fsp=True)._translate_default_for_mysql("TIME(2)", "time('now')") + == "CURRENT_TIME(2)" + ) + assert ( + self._mk(expr=True, ts_dt=False, fsp=False)._translate_default_for_mysql("TIME(2)", "time('now')") + == "CURRENT_TIME" + ) From e4963dcbc1bd9a66bca8b80d8b2617822b2e7ed4 Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Thu, 18 Sep 2025 21:30:18 +0100 Subject: [PATCH 09/17] :white_check_mark: add tests for MySQL expression defaults, CURRENT_TIMESTAMP, and fractional seconds support --- tests/unit/mysql_utils_test.py | 82 ++++++++++++++++++++++++++++++++++ 1 file changed, 82 insertions(+) diff --git a/tests/unit/mysql_utils_test.py b/tests/unit/mysql_utils_test.py index 53c5a92..c6743af 100644 --- a/tests/unit/mysql_utils_test.py +++ b/tests/unit/mysql_utils_test.py @@ -5,6 +5,9 @@ from sqlite3_to_mysql.mysql_utils import ( CharSet, + check_mysql_current_timestamp_datetime_support, + check_mysql_expression_defaults_support, + check_mysql_fractional_seconds_support, check_mysql_fulltext_support, check_mysql_json_support, check_mysql_values_alias_support, @@ -208,3 +211,82 @@ def __getitem__(self, key): result = list(mysql_supported_character_sets(charset="utf8")) # The function should skip the KeyError and return an empty list assert len(result) == 0 + + # ----------------------------- + # Expression defaults (MySQL 8.0.13+, MariaDB 10.2.0+) + # ----------------------------- + @pytest.mark.parametrize( + "ver, expected", + [ + ("8.0.12", False), + ("8.0.13", True), + ("8.0.13-8ubuntu1", True), + ("5.7.44", False), + ], + ) + def test_expr_defaults_mysql(self, ver: str, expected: bool) -> None: + assert check_mysql_expression_defaults_support(ver) is expected + + @pytest.mark.parametrize( + "ver, expected", + [ + ("10.1.99-MariaDB", False), + ("10.2.0-MariaDB", True), + ("10.2.7-MariaDB-1~deb10u1", True), + ("10.1.2-mArIaDb", False), # case-insensitive detection + ], + ) + def test_expr_defaults_mariadb(self, ver: str, expected: bool) -> None: + assert check_mysql_expression_defaults_support(ver) is expected + + # ----------------------------- + # CURRENT_TIMESTAMP for DATETIME (MySQL 5.6.5+, MariaDB 10.0.1+) + # ----------------------------- + @pytest.mark.parametrize( + "ver, expected", + [ + ("5.6.4", False), + ("5.6.5", True), + ("5.6.5-ps-log", True), + ("5.5.62", False), + ], + ) + def test_current_timestamp_datetime_mysql(self, ver: str, expected: bool) -> None: + assert check_mysql_current_timestamp_datetime_support(ver) is expected + + @pytest.mark.parametrize( + "ver, expected", + [ + ("10.0.0-MariaDB", False), + ("10.0.1-MariaDB", True), + ("10.3.39-MariaDB-1:10.3.39+maria~focal", True), + ], + ) + def test_current_timestamp_datetime_mariadb(self, ver: str, expected: bool) -> None: + assert check_mysql_current_timestamp_datetime_support(ver) is expected + + # ----------------------------- + # Fractional seconds (fsp) (MySQL 5.6.4+, MariaDB 10.1.2+) + # ----------------------------- + @pytest.mark.parametrize( + "ver, expected", + [ + ("5.6.3", False), + ("5.6.4", True), + ("5.7.44-0ubuntu0.18.04.1", True), + ], + ) + def test_fractional_seconds_mysql(self, ver: str, expected: bool) -> None: + assert check_mysql_fractional_seconds_support(ver) is expected + + @pytest.mark.parametrize( + "ver, expected", + [ + ("10.1.1-MariaDB", False), + ("10.1.2-MariaDB", True), + ("10.6.16-MariaDB-1:10.6.16+maria~jammy", True), + ("10.1.2-mArIaDb", True), # case-insensitive detection + ], + ) + def test_fractional_seconds_mariadb(self, ver: str, expected: bool) -> None: + assert check_mysql_fractional_seconds_support(ver) is expected From 791d3da70e1936f4d75e274a79cd49479f200199 Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Thu, 18 Sep 2025 21:35:25 +0100 Subject: [PATCH 10/17] :bug: handle falsy default values correctly in MySQL translation --- src/sqlite3_to_mysql/transporter.py | 20 ++++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/src/sqlite3_to_mysql/transporter.py b/src/sqlite3_to_mysql/transporter.py index 0e16438..f9c9815 100644 --- a/src/sqlite3_to_mysql/transporter.py +++ b/src/sqlite3_to_mysql/transporter.py @@ -490,7 +490,7 @@ def _column_type_length(cls, column_type: str, default: t.Optional[t.Union[str, suffix: t.Optional[t.Match[str]] = cls.COLUMN_LENGTH_PATTERN.search(column_type) if suffix: return suffix.group(0) - if default: + if default is not None: return f"({default})" return "" @@ -532,18 +532,22 @@ def _create_table(self, table_name: str, transfer_rowid: bool = False) -> None: column["pk"] > 0 and column_type.startswith(("INT", "BIGINT")) and not compound_primary_key ) + # Build DEFAULT clause safely (preserve falsy defaults like 0/'') + default_clause: str = "" + if ( + column["dflt_value"] is not None + and column_type not in MYSQL_COLUMN_TYPES_WITHOUT_DEFAULT + and not auto_increment + ): + td: str = self._translate_default_for_mysql(column_type, str(column["dflt_value"])) + if td != "": + default_clause = "DEFAULT " + td sql += " `{name}` {type} {notnull} {default} {auto_increment}, ".format( name=mysql_safe_name, type=column_type, notnull="NOT NULL" if column["notnull"] or column["pk"] else "NULL", auto_increment="AUTO_INCREMENT" if auto_increment else "", - default=( - "DEFAULT " + self._translate_default_for_mysql(column_type, str(column["dflt_value"])) - if column["dflt_value"] - and column_type not in MYSQL_COLUMN_TYPES_WITHOUT_DEFAULT - and not auto_increment - else "" - ), + default=default_clause, ) if column["pk"] > 0: From b08a37c14ea37998bc9d66a2639f5d5ee04d0a3b Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Thu, 18 Sep 2025 22:24:46 +0100 Subject: [PATCH 11/17] :sparkles: enhance TIMESTAMP and DATETIME handling for CURRENT_TIMESTAMP with fractional seconds support --- src/sqlite3_to_mysql/transporter.py | 31 ++++++++++++++++++++++------- 1 file changed, 24 insertions(+), 7 deletions(-) diff --git a/src/sqlite3_to_mysql/transporter.py b/src/sqlite3_to_mysql/transporter.py index f9c9815..3b2b9c0 100644 --- a/src/sqlite3_to_mysql/transporter.py +++ b/src/sqlite3_to_mysql/transporter.py @@ -411,16 +411,33 @@ def _translate_default_for_mysql(self, column_type: str, default: str) -> str: match: t.Optional[re.Match[str]] = self._valid_column_type(column_type) base: str = match.group(0).upper() if match else column_type.upper() - # TIMESTAMP/DATETIME - if ( - (base.startswith("TIMESTAMP") or base.startswith("DATETIME")) - and (self.CURRENT_TS.match(s) or self.SQLITE_NOW_FUNC.match(s) or self.STRFTIME_NOW.match(s)) - and self._allow_current_ts_dt + # TIMESTAMP: allow CURRENT_TIMESTAMP across versions; preserve FSP only if supported + if base.startswith("TIMESTAMP") and ( + self.CURRENT_TS.match(s) + or (self.SQLITE_NOW_FUNC.match(s) and s.lower().startswith("datetime")) + or self.STRFTIME_NOW.match(s) ): - # Too old for CURRENT_TIMESTAMP defaults on DATETIME/TIMESTAMP → fall back len_match: t.Optional[re.Match[str]] = self.COLUMN_LENGTH_PATTERN.search(column_type) fsp: str = "" - n: t.Optional[int] + if self._allow_fsp and len_match: + try: + n = int(len_match.group(0).strip("()")) + except ValueError: + n = None + if n is not None and 0 < n <= 6: + fsp = f"({n})" + return f"CURRENT_TIMESTAMP{fsp}" + + # DATETIME: require server support, otherwise omit the DEFAULT + if base.startswith("DATETIME") and ( + self.CURRENT_TS.match(s) + or (self.SQLITE_NOW_FUNC.match(s) and s.lower().startswith("datetime")) + or self.STRFTIME_NOW.match(s) + ): + if not self._allow_current_ts_dt: + return "" + len_match = self.COLUMN_LENGTH_PATTERN.search(column_type) + fsp = "" if self._allow_fsp and len_match: try: n = int(len_match.group(0).strip("()")) From efcef29910a696970ea109057929f1d7ef268f3f Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Thu, 18 Sep 2025 22:24:51 +0100 Subject: [PATCH 12/17] :white_check_mark: mark tests for missing database name and user as expected failures --- tests/func/test_cli.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/func/test_cli.py b/tests/func/test_cli.py index aa5cc08..a80b999 100644 --- a/tests/func/test_cli.py +++ b/tests/func/test_cli.py @@ -36,6 +36,7 @@ def test_non_existing_sqlite_file(self, cli_runner: CliRunner, mysql_database: E assert "Error: Invalid value" in result.output assert "does not exist" in result.output + @pytest.mark.xfail def test_no_database_name(self, cli_runner: CliRunner, sqlite_database: str, mysql_database: Engine) -> None: result = cli_runner.invoke(sqlite3mysql, ["-f", sqlite_database]) assert result.exit_code > 0 @@ -47,6 +48,7 @@ def test_no_database_name(self, cli_runner: CliRunner, sqlite_database: str, mys } ) + @pytest.mark.xfail def test_no_database_user( self, cli_runner: CliRunner, sqlite_database: str, mysql_credentials: MySQLCredentials, mysql_database: Engine ) -> None: From a0a528f94d523b18a58177508bf9a866ab583d1a Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Thu, 18 Sep 2025 22:25:04 +0100 Subject: [PATCH 13/17] :white_check_mark: refactor test cases for translating SQLite default values to MySQL for clarity and consistency --- tests/unit/sqlite3_to_mysql_test.py | 56 ++++++++++++++--------------- 1 file changed, 28 insertions(+), 28 deletions(-) diff --git a/tests/unit/sqlite3_to_mysql_test.py b/tests/unit/sqlite3_to_mysql_test.py index 7191599..c1dcec3 100644 --- a/tests/unit/sqlite3_to_mysql_test.py +++ b/tests/unit/sqlite3_to_mysql_test.py @@ -659,53 +659,53 @@ def test_strip_wrapping_parentheses(self, expr: str, expected: str) -> None: def _mk(*, expr: bool, ts_dt: bool, fsp: bool) -> SQLite3toMySQL: """ Build a lightweight instance without hitting __init__ (no DB connection needed). - Toggle the same feature flags transporter sets after version checks. + Toggle the same feature flags transporter§ sets after version checks. """ - o = SQLite3toMySQL.__new__(SQLite3toMySQL) - o._allow_expr_defaults = expr # MySQL >= 8.0.13 - o._allow_current_ts_dt = ts_dt # MySQL >= 5.6.5 - o._allow_fsp = fsp # MySQL >= 5.6.4 - return o + instance: SQLite3toMySQL = SQLite3toMySQL.__new__(SQLite3toMySQL) + instance._allow_expr_defaults = expr # MySQL >= 8.0.13 + instance._allow_current_ts_dt = ts_dt # MySQL >= 5.6.5 + instance._allow_fsp = fsp # MySQL >= 5.6.4 + return instance @pytest.mark.parametrize( "col, default, flags, expected", [ # --- TIMESTAMP/DATETIME + CURRENT_TIMESTAMP / now() mapping --- # Too old for CURRENT_TIMESTAMP on TIMESTAMP: fall back to stripped expr - ("TIMESTAMP(3)", "CURRENT_TIMESTAMP", dict(expr=False, ts_dt=False, fsp=False), "CURRENT_TIMESTAMP"), + ("TIMESTAMP(3)", "CURRENT_TIMESTAMP", {"expr": False, "ts_dt": False, "fsp": False}, "CURRENT_TIMESTAMP"), # Allowed, but no FSP support - ("TIMESTAMP(3)", "CURRENT_TIMESTAMP", dict(expr=False, ts_dt=True, fsp=False), "CURRENT_TIMESTAMP"), + ("TIMESTAMP(3)", "CURRENT_TIMESTAMP", {"expr": False, "ts_dt": True, "fsp": False}, "CURRENT_TIMESTAMP"), # Allowed with FSP support -> keep precision - ("TIMESTAMP(3)", "CURRENT_TIMESTAMP", dict(expr=False, ts_dt=True, fsp=True), "CURRENT_TIMESTAMP(3)"), + ("TIMESTAMP(3)", "CURRENT_TIMESTAMP", {"expr": False, "ts_dt": True, "fsp": True}, "CURRENT_TIMESTAMP(3)"), # SQLite-style now -> map to CURRENT_TIMESTAMP (with FSP when allowed) - ("DATETIME(2)", "datetime('now')", dict(expr=False, ts_dt=True, fsp=True), "CURRENT_TIMESTAMP(2)"), + ("DATETIME(2)", "datetime('now')", {"expr": False, "ts_dt": True, "fsp": True}, "CURRENT_TIMESTAMP(2)"), # --- DATE mapping (from 'now' forms or CURRENT_TIMESTAMP) --- # Only map when expression defaults are allowed - ("DATE", "datetime('now')", dict(expr=True, ts_dt=False, fsp=False), "CURRENT_DATE"), - ("DATE", "datetime('now')", dict(expr=False, ts_dt=False, fsp=False), "datetime('now')"), - ("DATE", "CURRENT_TIMESTAMP", dict(expr=True, ts_dt=True, fsp=True), "CURRENT_DATE"), - ("DATE", "CURRENT_TIMESTAMP", dict(expr=False, ts_dt=True, fsp=True), "CURRENT_TIMESTAMP"), + ("DATE", "datetime('now')", {"expr": True, "ts_dt": False, "fsp": False}, "CURRENT_DATE"), + ("DATE", "datetime('now')", {"expr": False, "ts_dt": False, "fsp": False}, "datetime('now')"), + ("DATE", "CURRENT_TIMESTAMP", {"expr": True, "ts_dt": True, "fsp": True}, "CURRENT_DATE"), + ("DATE", "CURRENT_TIMESTAMP", {"expr": False, "ts_dt": True, "fsp": True}, "CURRENT_TIMESTAMP"), # --- TIME mapping (from 'now' forms or CURRENT_TIMESTAMP) --- - ("TIME(3)", "CURRENT_TIME", dict(expr=True, ts_dt=False, fsp=True), "CURRENT_TIME(3)"), - ("TIME(3)", "CURRENT_TIME", dict(expr=True, ts_dt=False, fsp=False), "CURRENT_TIME"), - ("TIME(6)", "CURRENT_TIMESTAMP", dict(expr=True, ts_dt=True, fsp=True), "CURRENT_TIME(6)"), - ("TIME(6)", "CURRENT_TIMESTAMP", dict(expr=False, ts_dt=True, fsp=True), "CURRENT_TIMESTAMP"), + ("TIME(3)", "CURRENT_TIME", {"expr": True, "ts_dt": False, "fsp": True}, "CURRENT_TIME(3)"), + ("TIME(3)", "CURRENT_TIME", {"expr": True, "ts_dt": False, "fsp": False}, "CURRENT_TIME"), + ("TIME(6)", "CURRENT_TIMESTAMP", {"expr": True, "ts_dt": True, "fsp": True}, "CURRENT_TIME(6)"), + ("TIME(6)", "CURRENT_TIMESTAMP", {"expr": False, "ts_dt": True, "fsp": True}, "CURRENT_TIMESTAMP"), # --- Boolean normalization (for BOOL/BOOLEAN/TINYINT) --- - ("BOOLEAN", "TRUE", dict(expr=False, ts_dt=False, fsp=False), "1"), - ("TINYINT(1)", "'FALSE'", dict(expr=False, ts_dt=False, fsp=False), "0"), + ("BOOLEAN", "TRUE", {"expr": False, "ts_dt": False, "fsp": False}, "1"), + ("TINYINT(1)", "'FALSE'", {"expr": False, "ts_dt": False, "fsp": False}, "0"), # --- Numeric literals (incl. scientific notation) --- - ("INT", "42", dict(expr=False, ts_dt=False, fsp=False), "42"), - ("DOUBLE", "-3.14", dict(expr=False, ts_dt=False, fsp=False), "-3.14"), - ("DOUBLE", "1e-3", dict(expr=False, ts_dt=False, fsp=False), "1e-3"), - ("DOUBLE", "-2.5E+10", dict(expr=False, ts_dt=False, fsp=False), "-2.5E+10"), + ("INT", "42", {"expr": False, "ts_dt": False, "fsp": False}, "42"), + ("DOUBLE", "-3.14", {"expr": False, "ts_dt": False, "fsp": False}, "-3.14"), + ("DOUBLE", "1e-3", {"expr": False, "ts_dt": False, "fsp": False}, "1e-3"), + ("DOUBLE", "-2.5E+10", {"expr": False, "ts_dt": False, "fsp": False}, "-2.5E+10"), # --- Quoted strings and hex blobs pass through unchanged --- - ("VARCHAR(10)", "'hello'", dict(expr=False, ts_dt=False, fsp=False), "'hello'"), - ("BLOB", "X'ABCD'", dict(expr=False, ts_dt=False, fsp=False), "X'ABCD'"), + ("VARCHAR(10)", "'hello'", {"expr": False, "ts_dt": False, "fsp": False}, "'hello'"), + ("BLOB", "X'ABCD'", {"expr": False, "ts_dt": False, "fsp": False}, "X'ABCD'"), # --- Expression fallback (strip fully wrapping parens, leave the expr) --- - ("VARCHAR(10)", "(1+2)", dict(expr=False, ts_dt=False, fsp=False), "1+2"), + ("VARCHAR(10)", "(1+2)", {"expr": False, "ts_dt": False, "fsp": False}, "1+2"), ], ) - def test_translate_default_for_mysql(self, col, default, flags, expected): + def test_translate_default_for_mysql(self, col: str, default: str, flags: t.Dict[str, bool], expected: str): assert self._mk(**flags)._translate_default_for_mysql(col, default) == expected def test_time_mapping_from_sqlite_now_respects_fsp(self): From a36c7dd176822b876c5b678bf39c58f9221ceb58 Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Thu, 18 Sep 2025 22:27:42 +0100 Subject: [PATCH 14/17] :bug: improve NUMERIC_LITERAL_PATTERN regex to handle leading decimal points --- src/sqlite3_to_mysql/transporter.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/sqlite3_to_mysql/transporter.py b/src/sqlite3_to_mysql/transporter.py index 3b2b9c0..7c0864b 100644 --- a/src/sqlite3_to_mysql/transporter.py +++ b/src/sqlite3_to_mysql/transporter.py @@ -73,7 +73,7 @@ class SQLite3toMySQL(SQLite3toMySQLAttributes): r"^strftime\s*\(\s*'([^']+)'\s*,\s*'now'(?:\s*,\s*'(localtime|utc)')?\s*\)$", re.IGNORECASE, ) - NUMERIC_LITERAL_PATTERN: t.Pattern[str] = re.compile(r"^[+-]?\d+(?:\.\d+)?(?:[eE][+-]?\d+)?$") + NUMERIC_LITERAL_PATTERN: t.Pattern[str] = re.compile(r"^[+-]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?$") MYSQL_CONNECTOR_VERSION: version.Version = version.parse(mysql_connector_version_string) From 1f461afdbbd8b2c78a8115e48973c43acaf4fe8d Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Thu, 18 Sep 2025 22:30:28 +0100 Subject: [PATCH 15/17] :bug: fix fallback behavior for expression defaults in MySQL translation --- src/sqlite3_to_mysql/transporter.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/sqlite3_to_mysql/transporter.py b/src/sqlite3_to_mysql/transporter.py index 7c0864b..e3fa5a8 100644 --- a/src/sqlite3_to_mysql/transporter.py +++ b/src/sqlite3_to_mysql/transporter.py @@ -500,7 +500,7 @@ def _translate_default_for_mysql(self, column_type: str, default: str) -> str: return s # Fallback: return stripped expression (MySQL 8.0.13+ allows expression defaults) - return s + return s if self._allow_expr_defaults else "" @classmethod def _column_type_length(cls, column_type: str, default: t.Optional[t.Union[str, int, float]] = None) -> str: From 4653fc420b9203f283cd2fd403050fb4ff663e85 Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Thu, 18 Sep 2025 22:30:32 +0100 Subject: [PATCH 16/17] :white_check_mark: update default value translations for DATETIME and DATE to return empty string for non-expressions --- tests/unit/sqlite3_to_mysql_test.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/tests/unit/sqlite3_to_mysql_test.py b/tests/unit/sqlite3_to_mysql_test.py index c1dcec3..2abcf2e 100644 --- a/tests/unit/sqlite3_to_mysql_test.py +++ b/tests/unit/sqlite3_to_mysql_test.py @@ -679,17 +679,18 @@ def _mk(*, expr: bool, ts_dt: bool, fsp: bool) -> SQLite3toMySQL: ("TIMESTAMP(3)", "CURRENT_TIMESTAMP", {"expr": False, "ts_dt": True, "fsp": True}, "CURRENT_TIMESTAMP(3)"), # SQLite-style now -> map to CURRENT_TIMESTAMP (with FSP when allowed) ("DATETIME(2)", "datetime('now')", {"expr": False, "ts_dt": True, "fsp": True}, "CURRENT_TIMESTAMP(2)"), + ("DATETIME(6)", "CURRENT_TIMESTAMP", dict(expr=False, ts_dt=False, fsp=True), ""), # --- DATE mapping (from 'now' forms or CURRENT_TIMESTAMP) --- # Only map when expression defaults are allowed ("DATE", "datetime('now')", {"expr": True, "ts_dt": False, "fsp": False}, "CURRENT_DATE"), - ("DATE", "datetime('now')", {"expr": False, "ts_dt": False, "fsp": False}, "datetime('now')"), + ("DATE", "datetime('now')", {"expr": False, "ts_dt": False, "fsp": False}, ""), ("DATE", "CURRENT_TIMESTAMP", {"expr": True, "ts_dt": True, "fsp": True}, "CURRENT_DATE"), - ("DATE", "CURRENT_TIMESTAMP", {"expr": False, "ts_dt": True, "fsp": True}, "CURRENT_TIMESTAMP"), + ("DATE", "CURRENT_TIMESTAMP", {"expr": False, "ts_dt": True, "fsp": True}, ""), # --- TIME mapping (from 'now' forms or CURRENT_TIMESTAMP) --- ("TIME(3)", "CURRENT_TIME", {"expr": True, "ts_dt": False, "fsp": True}, "CURRENT_TIME(3)"), ("TIME(3)", "CURRENT_TIME", {"expr": True, "ts_dt": False, "fsp": False}, "CURRENT_TIME"), ("TIME(6)", "CURRENT_TIMESTAMP", {"expr": True, "ts_dt": True, "fsp": True}, "CURRENT_TIME(6)"), - ("TIME(6)", "CURRENT_TIMESTAMP", {"expr": False, "ts_dt": True, "fsp": True}, "CURRENT_TIMESTAMP"), + ("TIME(6)", "CURRENT_TIMESTAMP", {"expr": False, "ts_dt": True, "fsp": True}, ""), # --- Boolean normalization (for BOOL/BOOLEAN/TINYINT) --- ("BOOLEAN", "TRUE", {"expr": False, "ts_dt": False, "fsp": False}, "1"), ("TINYINT(1)", "'FALSE'", {"expr": False, "ts_dt": False, "fsp": False}, "0"), @@ -702,7 +703,7 @@ def _mk(*, expr: bool, ts_dt: bool, fsp: bool) -> SQLite3toMySQL: ("VARCHAR(10)", "'hello'", {"expr": False, "ts_dt": False, "fsp": False}, "'hello'"), ("BLOB", "X'ABCD'", {"expr": False, "ts_dt": False, "fsp": False}, "X'ABCD'"), # --- Expression fallback (strip fully wrapping parens, leave the expr) --- - ("VARCHAR(10)", "(1+2)", {"expr": False, "ts_dt": False, "fsp": False}, "1+2"), + ("VARCHAR(10)", "(1+2)", {"expr": False, "ts_dt": False, "fsp": False}, ""), ], ) def test_translate_default_for_mysql(self, col: str, default: str, flags: t.Dict[str, bool], expected: str): From b7bd38e93ee27e579af3a736c972f09a27cd8f0f Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Fri, 19 Sep 2025 09:01:59 +0100 Subject: [PATCH 17/17] :rewind: update default value translations for DATE and TIME to return expressions instead of empty strings --- src/sqlite3_to_mysql/transporter.py | 2 +- tests/unit/sqlite3_to_mysql_test.py | 9 ++++----- 2 files changed, 5 insertions(+), 6 deletions(-) diff --git a/src/sqlite3_to_mysql/transporter.py b/src/sqlite3_to_mysql/transporter.py index e3fa5a8..7c0864b 100644 --- a/src/sqlite3_to_mysql/transporter.py +++ b/src/sqlite3_to_mysql/transporter.py @@ -500,7 +500,7 @@ def _translate_default_for_mysql(self, column_type: str, default: str) -> str: return s # Fallback: return stripped expression (MySQL 8.0.13+ allows expression defaults) - return s if self._allow_expr_defaults else "" + return s @classmethod def _column_type_length(cls, column_type: str, default: t.Optional[t.Union[str, int, float]] = None) -> str: diff --git a/tests/unit/sqlite3_to_mysql_test.py b/tests/unit/sqlite3_to_mysql_test.py index 2abcf2e..c1dcec3 100644 --- a/tests/unit/sqlite3_to_mysql_test.py +++ b/tests/unit/sqlite3_to_mysql_test.py @@ -679,18 +679,17 @@ def _mk(*, expr: bool, ts_dt: bool, fsp: bool) -> SQLite3toMySQL: ("TIMESTAMP(3)", "CURRENT_TIMESTAMP", {"expr": False, "ts_dt": True, "fsp": True}, "CURRENT_TIMESTAMP(3)"), # SQLite-style now -> map to CURRENT_TIMESTAMP (with FSP when allowed) ("DATETIME(2)", "datetime('now')", {"expr": False, "ts_dt": True, "fsp": True}, "CURRENT_TIMESTAMP(2)"), - ("DATETIME(6)", "CURRENT_TIMESTAMP", dict(expr=False, ts_dt=False, fsp=True), ""), # --- DATE mapping (from 'now' forms or CURRENT_TIMESTAMP) --- # Only map when expression defaults are allowed ("DATE", "datetime('now')", {"expr": True, "ts_dt": False, "fsp": False}, "CURRENT_DATE"), - ("DATE", "datetime('now')", {"expr": False, "ts_dt": False, "fsp": False}, ""), + ("DATE", "datetime('now')", {"expr": False, "ts_dt": False, "fsp": False}, "datetime('now')"), ("DATE", "CURRENT_TIMESTAMP", {"expr": True, "ts_dt": True, "fsp": True}, "CURRENT_DATE"), - ("DATE", "CURRENT_TIMESTAMP", {"expr": False, "ts_dt": True, "fsp": True}, ""), + ("DATE", "CURRENT_TIMESTAMP", {"expr": False, "ts_dt": True, "fsp": True}, "CURRENT_TIMESTAMP"), # --- TIME mapping (from 'now' forms or CURRENT_TIMESTAMP) --- ("TIME(3)", "CURRENT_TIME", {"expr": True, "ts_dt": False, "fsp": True}, "CURRENT_TIME(3)"), ("TIME(3)", "CURRENT_TIME", {"expr": True, "ts_dt": False, "fsp": False}, "CURRENT_TIME"), ("TIME(6)", "CURRENT_TIMESTAMP", {"expr": True, "ts_dt": True, "fsp": True}, "CURRENT_TIME(6)"), - ("TIME(6)", "CURRENT_TIMESTAMP", {"expr": False, "ts_dt": True, "fsp": True}, ""), + ("TIME(6)", "CURRENT_TIMESTAMP", {"expr": False, "ts_dt": True, "fsp": True}, "CURRENT_TIMESTAMP"), # --- Boolean normalization (for BOOL/BOOLEAN/TINYINT) --- ("BOOLEAN", "TRUE", {"expr": False, "ts_dt": False, "fsp": False}, "1"), ("TINYINT(1)", "'FALSE'", {"expr": False, "ts_dt": False, "fsp": False}, "0"), @@ -703,7 +702,7 @@ def _mk(*, expr: bool, ts_dt: bool, fsp: bool) -> SQLite3toMySQL: ("VARCHAR(10)", "'hello'", {"expr": False, "ts_dt": False, "fsp": False}, "'hello'"), ("BLOB", "X'ABCD'", {"expr": False, "ts_dt": False, "fsp": False}, "X'ABCD'"), # --- Expression fallback (strip fully wrapping parens, leave the expr) --- - ("VARCHAR(10)", "(1+2)", {"expr": False, "ts_dt": False, "fsp": False}, ""), + ("VARCHAR(10)", "(1+2)", {"expr": False, "ts_dt": False, "fsp": False}, "1+2"), ], ) def test_translate_default_for_mysql(self, col: str, default: str, flags: t.Dict[str, bool], expected: str):