From 55e0198b9f80372a7618daf1fb97e46f6614c9c2 Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Sun, 16 Nov 2025 16:13:13 +0000 Subject: [PATCH 1/3] :sparkles: add MySQL table prefix option with validation in CLI --- src/sqlite3_to_mysql/cli.py | 22 ++++++++ src/sqlite3_to_mysql/transporter.py | 86 ++++++++++++++++++----------- src/sqlite3_to_mysql/types.py | 2 + 3 files changed, 78 insertions(+), 32 deletions(-) diff --git a/src/sqlite3_to_mysql/cli.py b/src/sqlite3_to_mysql/cli.py index fbcc5eb..056570d 100644 --- a/src/sqlite3_to_mysql/cli.py +++ b/src/sqlite3_to_mysql/cli.py @@ -14,11 +14,24 @@ from .click_utils import OptionEatAll, prompt_password from .debug_info import info from .mysql_utils import MYSQL_INSERT_METHOD, MYSQL_TEXT_COLUMN_TYPES, mysql_supported_character_sets +from .transporter import MYSQL_TABLE_PREFIX_PATTERN _copyright_header: str = f"sqlite3mysql version {package_version} Copyright (c) 2018-{datetime.now().year} Klemen Tusar" +def _validate_mysql_table_prefix(_: t.Any, __: t.Any, value: t.Optional[str]) -> str: + """Validate the optional MySQL table prefix supplied via CLI.""" + if not value: + return "" + if not MYSQL_TABLE_PREFIX_PATTERN.match(value): + raise click.BadParameter( + "Table prefix must start with a letter, contain only letters, numbers, or underscores, " + "and be at most 32 characters long." + ) + return value + + @click.command( name="sqlite3mysql", help=_copyright_header, @@ -116,6 +129,13 @@ default="TEXT", help="MySQL default text field type. Defaults to TEXT.", ) +@click.option( + "-b", + "--mysql-table-prefix", + default="", + callback=_validate_mysql_table_prefix, + help="Prefix to prepend to every created MySQL table (letters/numbers/underscores, must start with a letter).", +) @click.option( "--mysql-charset", metavar="TEXT", @@ -169,6 +189,7 @@ def cli( mysql_integer_type: str, mysql_string_type: str, mysql_text_type: str, + mysql_table_prefix: str, mysql_charset: str, mysql_collation: str, use_fulltext: bool, @@ -224,6 +245,7 @@ def cli( mysql_integer_type=mysql_integer_type, mysql_string_type=mysql_string_type, mysql_text_type=mysql_text_type, + mysql_table_prefix=mysql_table_prefix, mysql_charset=mysql_charset.lower() if mysql_charset else "utf8mb4", mysql_collation=mysql_collation.lower() if mysql_collation else None, ignore_duplicate_keys=ignore_duplicate_keys, diff --git a/src/sqlite3_to_mysql/transporter.py b/src/sqlite3_to_mysql/transporter.py index a7cc50f..507b8b4 100644 --- a/src/sqlite3_to_mysql/transporter.py +++ b/src/sqlite3_to_mysql/transporter.py @@ -66,6 +66,7 @@ key: value for key, value in sqlglot_mysql.MySQL.INVERSE_TIME_MAPPING.items() if key != "%H:%M:%S" } SQLGLOT_MYSQL_INVERSE_TIME_TRIE: t.Dict[str, t.Any] = new_trie(SQLGLOT_MYSQL_INVERSE_TIME_MAPPING) +MYSQL_TABLE_PREFIX_PATTERN: t.Pattern[str] = re.compile(r"^[A-Za-z][A-Za-z0-9_]{0,31}$") class SQLite3toMySQL(SQLite3toMySQLAttributes): @@ -91,6 +92,7 @@ class SQLite3toMySQL(SQLite3toMySQLAttributes): re.IGNORECASE, ) NUMERIC_LITERAL_PATTERN: t.Pattern[str] = re.compile(r"^[+-]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?$") + TABLE_PREFIX_PATTERN: t.Pattern[str] = MYSQL_TABLE_PREFIX_PATTERN MYSQL_CONNECTOR_VERSION: version.Version = version.parse(mysql_connector_version_string) @@ -174,6 +176,14 @@ def __init__(self, **kwargs: Unpack[SQLite3toMySQLParams]): if not kwargs.get("mysql_collation") and self._mysql_collation == "utf8mb4_0900_ai_ci": self._mysql_collation = "utf8mb4_unicode_ci" + mysql_table_prefix: str = str(kwargs.get("mysql_table_prefix", "") or "") + if mysql_table_prefix and not self.TABLE_PREFIX_PATTERN.match(mysql_table_prefix): + raise ValueError( + "MySQL table prefix must start with a letter and contain only letters, numbers, or underscores " + "with a maximum length of 32 characters." + ) + self._mysql_table_prefix = mysql_table_prefix + self._ignore_duplicate_keys = kwargs.get("ignore_duplicate_keys", False) or False self._use_fulltext = kwargs.get("use_fulltext", False) or False @@ -339,6 +349,13 @@ def _sqlite_table_has_rowid(self, table: str) -> bool: except sqlite3.OperationalError: return False + def _mysql_table_name(self, table_name: str) -> str: + """Return the MySQL table name with any configured prefix applied.""" + prefix: str = getattr(self, "_mysql_table_prefix", "") + if prefix: + return safe_identifier_length(f"{prefix}{table_name}") + return safe_identifier_length(table_name) + def _create_database(self) -> None: try: self._mysql_cur.execute( @@ -881,8 +898,9 @@ def _create_mysql_view(self, view_name: str, view_sql: str) -> None: def _create_table(self, table_name: str, transfer_rowid: bool = False, skip_default: bool = False) -> None: primary_keys: t.List[t.Dict[str, str]] = [] + mysql_table_name: str = self._mysql_table_name(table_name) - sql: str = f"CREATE TABLE IF NOT EXISTS `{safe_identifier_length(table_name)}` ( " + sql: str = f"CREATE TABLE IF NOT EXISTS `{mysql_table_name}` ( " if transfer_rowid: sql += " `rowid` BIGINT NOT NULL, " @@ -960,7 +978,7 @@ def _create_table(self, table_name: str, transfer_rowid: bool = False, skip_defa ) if transfer_rowid: - sql += f", CONSTRAINT `{safe_identifier_length(table_name)}_rowid` UNIQUE (`rowid`)" + sql += f", CONSTRAINT `{mysql_table_name}_rowid` UNIQUE (`rowid`)" sql += f" ) ENGINE=InnoDB DEFAULT CHARSET={self._mysql_charset} COLLATE={self._mysql_collation}" @@ -971,19 +989,20 @@ def _create_table(self, table_name: str, transfer_rowid: bool = False, skip_defa if err.errno == errorcode.ER_INVALID_DEFAULT and not skip_default: self._logger.warning( "MySQL failed creating table %s with DEFAULT values: %s. Retrying without DEFAULT values ...", - safe_identifier_length(table_name), + mysql_table_name, err, ) return self._create_table(table_name, transfer_rowid, skip_default=True) else: self._logger.error( "MySQL failed creating table %s: %s", - safe_identifier_length(table_name), + mysql_table_name, err, ) raise def _truncate_table(self, table_name: str) -> None: + mysql_table_name: str = self._mysql_table_name(table_name) self._mysql_cur.execute( """ SELECT `TABLE_NAME` @@ -992,14 +1011,15 @@ def _truncate_table(self, table_name: str) -> None: AND `TABLE_NAME` = %s LIMIT 1 """, - (self._mysql_database, safe_identifier_length(table_name)), + (self._mysql_database, mysql_table_name), ) if len(self._mysql_cur.fetchall()) > 0: - self._logger.info("Truncating table %s", safe_identifier_length(table_name)) - self._mysql_cur.execute(f"TRUNCATE TABLE `{safe_identifier_length(table_name)}`") + self._logger.info("Truncating table %s", mysql_table_name) + self._mysql_cur.execute(f"TRUNCATE TABLE `{mysql_table_name}`") def _add_indices(self, table_name: str) -> None: quoted_table_name: str = self._sqlite_quote_ident(table_name) + mysql_table_name: str = self._mysql_table_name(table_name) self._sqlite_cur.execute(f'PRAGMA table_info("{quoted_table_name}")') table_columns: t.Dict[str, str] = {} @@ -1063,7 +1083,7 @@ def _add_indices(self, table_name: str) -> None: self._logger.warning( """Failed adding index to column "%s" in table %s: Column not found!""", ", ".join(safe_identifier_length(index_info["name"]) for index_info in index_infos), - safe_identifier_length(table_name), + mysql_table_name, ) continue @@ -1107,12 +1127,13 @@ def _add_index( index_infos: t.Tuple[t.Dict[str, t.Any], ...], index_iteration: int = 0, ) -> None: + mysql_table_name: str = self._mysql_table_name(table_name) sql: str = ( """ ALTER TABLE `{table}` ADD {index_type} `{name}`({columns}) """.format( - table=safe_identifier_length(table_name), + table=mysql_table_name, index_type=index_type, name=( safe_identifier_length(index["name"]) @@ -1128,14 +1149,13 @@ def _add_index( """Adding %s to column "%s" in table %s""", "unique index" if int(index["unique"]) == 1 else "index", ", ".join(safe_identifier_length(index_info["name"]) for index_info in index_infos), - safe_identifier_length(table_name), + mysql_table_name, ) self._mysql_cur.execute(sql) self._mysql.commit() except mysql.connector.Error as err: if err.errno == errorcode.ER_DUP_KEYNAME: if not self._ignore_duplicate_keys: - # handle a duplicate key name self._add_index( table_name=table_name, index_type=index_type, @@ -1147,7 +1167,7 @@ def _add_index( self._logger.warning( """Duplicate key "%s" in table %s detected! Trying to create new key "%s_%s" ...""", safe_identifier_length(index["name"]), - safe_identifier_length(table_name), + mysql_table_name, safe_identifier_length(index["name"]), index_iteration + 1, ) @@ -1155,51 +1175,51 @@ def _add_index( self._logger.warning( """Ignoring duplicate key "%s" in table %s!""", safe_identifier_length(index["name"]), - safe_identifier_length(table_name), + mysql_table_name, ) elif err.errno == errorcode.ER_DUP_ENTRY: self._logger.warning( """Ignoring duplicate entry when adding index to column "%s" in table %s!""", ", ".join(safe_identifier_length(index_info["name"]) for index_info in index_infos), - safe_identifier_length(table_name), + mysql_table_name, ) elif err.errno == errorcode.ER_DUP_FIELDNAME: self._logger.warning( """Failed adding index to column "%s" in table %s: Duplicate field name! Ignoring...""", ", ".join(safe_identifier_length(index_info["name"]) for index_info in index_infos), - safe_identifier_length(table_name), + mysql_table_name, ) elif err.errno == errorcode.ER_TOO_MANY_KEYS: self._logger.warning( """Failed adding index to column "%s" in table %s: Too many keys! Ignoring...""", ", ".join(safe_identifier_length(index_info["name"]) for index_info in index_infos), - safe_identifier_length(table_name), + mysql_table_name, ) elif err.errno == errorcode.ER_TOO_LONG_KEY: self._logger.warning( """Failed adding index to column "%s" in table %s: Key length too long! Ignoring...""", ", ".join(safe_identifier_length(index_info["name"]) for index_info in index_infos), - safe_identifier_length(table_name), + mysql_table_name, ) elif err.errno == errorcode.ER_BAD_FT_COLUMN: - # handle bad FULLTEXT index self._logger.warning( """Failed adding FULLTEXT index to column "%s" in table %s. Retrying without FULLTEXT ...""", ", ".join(safe_identifier_length(index_info["name"]) for index_info in index_infos), - safe_identifier_length(table_name), + mysql_table_name, ) raise else: self._logger.error( """MySQL failed adding index to column "%s" in table %s: %s""", ", ".join(safe_identifier_length(index_info["name"]) for index_info in index_infos), - safe_identifier_length(table_name), + mysql_table_name, err, ) raise def _add_foreign_keys(self, table_name: str) -> None: quoted_table_name: str = self._sqlite_quote_ident(table_name) + mysql_table_name: str = self._mysql_table_name(table_name) self._sqlite_cur.execute(f'PRAGMA foreign_key_list("{quoted_table_name}")') foreign_keys: t.Dict[int, t.List[t.Dict[str, t.Any]]] = {} @@ -1219,7 +1239,7 @@ def _add_foreign_keys(self, table_name: str) -> None: self._logger.warning( 'Skipping foreign key "%s" in table %s: partially defined reference columns.', safe_identifier_length(fk_rows[0]["from"]), - safe_identifier_length(table_name), + mysql_table_name, ) continue @@ -1228,14 +1248,15 @@ def _add_foreign_keys(self, table_name: str) -> None: self._logger.warning( 'Skipping foreign key "%s" in table %s: unable to resolve referenced primary key columns from table %s.', safe_identifier_length(fk_rows[0]["from"]), - safe_identifier_length(table_name), - safe_identifier_length(ref_table), + mysql_table_name, + self._mysql_table_name(ref_table), ) continue referenced_columns = primary_keys else: referenced_columns = [safe_identifier_length(fk_row["to"]) for fk_row in fk_rows] + mysql_ref_table_name: str = self._mysql_table_name(ref_table) sql = """ ALTER TABLE `{table}` ADD CONSTRAINT `{table}_FK_{id}_{seq}` @@ -1246,9 +1267,9 @@ def _add_foreign_keys(self, table_name: str) -> None: """.format( id=fk_id, seq=fk_rows[0]["seq"], - table=safe_identifier_length(table_name), + table=mysql_table_name, columns=", ".join(f"`{column}`" for column in from_columns), - ref_table=safe_identifier_length(ref_table), + ref_table=mysql_ref_table_name, ref_columns=", ".join(f"`{column}`" for column in referenced_columns), on_delete=( fk_rows[0]["on_delete"].upper() if fk_rows[0]["on_delete"].upper() != "SET DEFAULT" else "NO ACTION" @@ -1261,9 +1282,9 @@ def _add_foreign_keys(self, table_name: str) -> None: try: self._logger.info( "Adding foreign key to %s.(%s) referencing %s.(%s)", - safe_identifier_length(table_name), + mysql_table_name, ", ".join(from_columns), - safe_identifier_length(ref_table), + mysql_ref_table_name, ", ".join(referenced_columns), ) self._mysql_cur.execute(sql) @@ -1271,9 +1292,9 @@ def _add_foreign_keys(self, table_name: str) -> None: except mysql.connector.Error as err: self._logger.error( "MySQL failed adding foreign key to %s.(%s) referencing %s.(%s): %s", - safe_identifier_length(table_name), + mysql_table_name, ", ".join(from_columns), - safe_identifier_length(ref_table), + mysql_ref_table_name, ", ".join(referenced_columns), err, ) @@ -1320,6 +1341,7 @@ def transfer(self) -> None: table_name: str = table["name"] object_type: str = table.get("type", "table") quoted_table_name: str = self._sqlite_quote_ident(table_name) + mysql_table_name: str = self._mysql_table_name(table_name) # check if we're transferring rowid transfer_rowid: bool = self._with_rowid and self._sqlite_table_has_rowid(table_name) @@ -1391,7 +1413,7 @@ def transfer(self) -> None: {values_clause} ON DUPLICATE KEY UPDATE {field_updates} """.format( - table=safe_identifier_length(table_name), + table=mysql_table_name, fields=("`{}`, " * len(columns)).rstrip(" ,").format(*columns), values_clause=( "VALUES ({placeholders}) AS `__new__`" @@ -1411,7 +1433,7 @@ def transfer(self) -> None: VALUES ({placeholders}) """.format( ignore="IGNORE" if self._mysql_insert_method.upper() == "IGNORE" else "", - table=safe_identifier_length(table_name), + table=mysql_table_name, fields=("`{}`, " * len(columns)).rstrip(" ,").format(*columns), placeholders=("%s, " * len(columns)).rstrip(" ,"), ) @@ -1421,7 +1443,7 @@ def transfer(self) -> None: self._logger.error( "MySQL transfer failed inserting data into %s %s: %s", "view" if object_type == "view" else "table", - safe_identifier_length(table_name), + mysql_table_name, err, ) raise diff --git a/src/sqlite3_to_mysql/types.py b/src/sqlite3_to_mysql/types.py index 6b74075..3cd6a82 100644 --- a/src/sqlite3_to_mysql/types.py +++ b/src/sqlite3_to_mysql/types.py @@ -47,6 +47,7 @@ class SQLite3toMySQLParams(TypedDict): mysql_insert_method: t.Optional[str] mysql_string_type: t.Optional[str] mysql_text_type: t.Optional[str] + mysql_table_prefix: t.Optional[str] class SQLite3toMySQLAttributes: @@ -75,6 +76,7 @@ class SQLite3toMySQLAttributes: _mysql_integer_type: str _mysql_string_type: str _mysql_text_type: str + _mysql_table_prefix: str _mysql_charset: str _mysql_collation: str _ignore_duplicate_keys: bool From 9a2d4cab4a62d8a048f8134a3cacbef43551dff7 Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Sun, 16 Nov 2025 16:13:30 +0000 Subject: [PATCH 2/3] :white_check_mark: add MySQL table prefix support and validation in CLI tests --- tests/unit/sqlite3_to_mysql_test.py | 109 ++++++++++++++++++++++++++++ 1 file changed, 109 insertions(+) diff --git a/tests/unit/sqlite3_to_mysql_test.py b/tests/unit/sqlite3_to_mysql_test.py index 28278ff..c42e696 100644 --- a/tests/unit/sqlite3_to_mysql_test.py +++ b/tests/unit/sqlite3_to_mysql_test.py @@ -67,6 +67,65 @@ def test_cli_sqlite_views_flag_propagates( assert transporter_ctor.call_args.kwargs["sqlite_views_as_tables"] is True +def test_cli_mysql_table_prefix_passed_to_transporter( + cli_runner: CliRunner, + sqlite_database: str, + mysql_credentials: MySQLCredentials, + mocker: MockerFixture, +) -> None: + transporter_ctor = mocker.patch("sqlite3_to_mysql.cli.SQLite3toMySQL", autospec=True) + transporter_instance = transporter_ctor.return_value + transporter_instance.transfer.return_value = None + + common_args = [ + "-f", + sqlite_database, + "-d", + mysql_credentials.database, + "-u", + mysql_credentials.user, + "--mysql-password", + mysql_credentials.password, + "-h", + mysql_credentials.host, + "-P", + str(mysql_credentials.port), + ] + + result: Result = cli_runner.invoke(sqlite3mysql, common_args + ["--mysql-table-prefix", "stage_"]) + assert result.exit_code == 0 + assert transporter_ctor.call_args.kwargs["mysql_table_prefix"] == "stage_" + + +def test_cli_mysql_table_prefix_validation( + cli_runner: CliRunner, + sqlite_database: str, + mysql_credentials: MySQLCredentials, + mocker: MockerFixture, +) -> None: + transporter_ctor = mocker.patch("sqlite3_to_mysql.cli.SQLite3toMySQL", autospec=True) + + common_args = [ + "-f", + sqlite_database, + "-d", + mysql_credentials.database, + "-u", + mysql_credentials.user, + "--mysql-password", + mysql_credentials.password, + "-h", + mysql_credentials.host, + "-P", + str(mysql_credentials.port), + ] + + result: Result = cli_runner.invoke(sqlite3mysql, common_args + ["--mysql-table-prefix", "123bad"]) + assert result.exit_code != 0 + assert "Table prefix" in result.output + transporter_ctor.assert_not_called() + + def test_cli_collation_validation( cli_runner: CliRunner, sqlite_database: str, @@ -401,6 +460,7 @@ def _make_transfer_stub(mocker: MockFixture) -> SQLite3toMySQL: instance._translate_sqlite_view_definition = mocker.MagicMock(return_value="CREATE VIEW translated AS SELECT 1") instance._sqlite_cur.fetchall.return_value = [] instance._sqlite_cur.execute.return_value = None + instance._mysql_table_prefix = "" instance._get_table_info = mocker.MagicMock( return_value=[ {"name": "c1", "type": "TEXT", "hidden": 0}, @@ -483,6 +543,27 @@ def execute_side_effect(sql, *params): assert "INSERT" in sql_arg +def test_transfer_applies_mysql_table_prefix(mocker: MockFixture) -> None: + instance = _make_transfer_stub(mocker) + instance._mysql_table_prefix = "pre_" + instance._mysql_transfer_data = True + instance._sqlite_cur.fetchone.return_value = {"total_records": 1} + instance._sqlite_cur.fetchall.return_value = [(1,)] + + def execute_side_effect(sql, *params): + if sql.startswith("SELECT ") and "FROM" in sql and "COUNT" not in sql.upper(): + instance._sqlite_cur.description = [("c1",)] + return None + + instance._sqlite_cur.execute.side_effect = execute_side_effect + instance._fetch_sqlite_master_rows = mocker.MagicMock(side_effect=[[{"name": "tbl", "type": "table"}], []]) + + instance.transfer() + + sql_arg = instance._transfer_table_data.call_args.kwargs["sql"] + assert "INTO `pre_tbl`" in sql_arg + + def test_transfer_escapes_sqlite_identifiers(mocker: MockFixture) -> None: instance = _make_transfer_stub(mocker) instance._mysql_transfer_data = True @@ -1717,6 +1798,34 @@ def test_add_foreign_keys_shorthand_references_primary_key( assert "REFERENCES `parent`(`id`)" in executed_sql proc._mysql.commit.assert_called_once() + def test_add_foreign_keys_apply_mysql_table_prefix(self, mocker: MockFixture) -> None: + proc = SQLite3toMySQL.__new__(SQLite3toMySQL) + sqlite_cursor = mocker.MagicMock() + sqlite_cursor.fetchall.return_value = [ + { + "id": 0, + "seq": 0, + "table": "parent", + "from": "parent_id", + "to": "id", + "on_delete": "NO ACTION", + "on_update": "NO ACTION", + } + ] + proc._sqlite_cur = sqlite_cursor + proc._sqlite_table_xinfo_support = False + proc._mysql_cur = mocker.MagicMock() + proc._mysql = mocker.MagicMock() + proc._logger = mocker.MagicMock() + proc._mysql_table_prefix = "pre_" + + proc._add_foreign_keys("child") + + executed_sql = proc._mysql_cur.execute.call_args[0][0] + assert "ALTER TABLE `pre_child`" in executed_sql + assert "REFERENCES `pre_parent`" in executed_sql + proc._mysql.commit.assert_called_once() + def test_add_foreign_keys_shorthand_pk_mismatch_is_skipped( self, sqlite_database: str, From e1291e510fdf2602e393455daa06121885699c0e Mon Sep 17 00:00:00 2001 From: Klemen Tusar Date: Sun, 16 Nov 2025 17:49:14 +0000 Subject: [PATCH 3/3] :bulb: enhance MySQL table prefix help message and clean up unused variables in transporter --- src/sqlite3_to_mysql/cli.py | 3 ++- src/sqlite3_to_mysql/transporter.py | 6 ++---- 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/src/sqlite3_to_mysql/cli.py b/src/sqlite3_to_mysql/cli.py index 056570d..d87da21 100644 --- a/src/sqlite3_to_mysql/cli.py +++ b/src/sqlite3_to_mysql/cli.py @@ -134,7 +134,8 @@ def _validate_mysql_table_prefix(_: t.Any, __: t.Any, value: t.Optional[str]) -> "--mysql-table-prefix", default="", callback=_validate_mysql_table_prefix, - help="Prefix to prepend to every created MySQL table (letters/numbers/underscores, must start with a letter).", + help="MySQL table prefix must start with a letter and contain only letters, numbers, or underscores " + "with a maximum length of 32 characters.", ) @click.option( "--mysql-charset", diff --git a/src/sqlite3_to_mysql/transporter.py b/src/sqlite3_to_mysql/transporter.py index 507b8b4..e990845 100644 --- a/src/sqlite3_to_mysql/transporter.py +++ b/src/sqlite3_to_mysql/transporter.py @@ -930,8 +930,6 @@ def _create_table(self, table_name: str, transfer_rowid: bool = False, skip_defa column["pk"] > 0 and column_type.startswith(("INT", "BIGINT")) and not compound_primary_key ) - allow_expr_defaults: bool = getattr(self, "_allow_expr_defaults", False) - is_mariadb: bool = getattr(self, "_is_mariadb", False) base_type: str = self._base_mysql_column_type(column_type) # Build DEFAULT clause safely (preserve falsy defaults like 0/'') @@ -939,14 +937,14 @@ def _create_table(self, table_name: str, transfer_rowid: bool = False, skip_defa if ( not skip_default and column["dflt_value"] is not None - and self._column_type_supports_default(base_type, allow_expr_defaults) + and self._column_type_supports_default(base_type, self._allow_expr_defaults) and not auto_increment ): td: str = self._translate_default_for_mysql(column_type, str(column["dflt_value"])) if td != "": stripped_td: str = td.strip() if base_type in MYSQL_TEXT_COLUMN_TYPES_WITH_JSON and stripped_td.upper() != "NULL": - td = self._format_textual_default(stripped_td, allow_expr_defaults, is_mariadb) + td = self._format_textual_default(stripped_td, self._allow_expr_defaults, self._is_mariadb) else: td = stripped_td default_clause = "DEFAULT " + td