⚡ use f-strings where appropriate

Author: techouse

mysql_to_sqlite3/click_utils.py

@@ -13,7 +13,7 @@ def __init__(self, *args, **kwargs):
         self.save_other_options = kwargs.pop("save_other_options", True)
         nargs = kwargs.pop("nargs", -1)
         if nargs != -1:
-            raise ValueError("nargs, if set, must be -1 not {}".format(nargs))
+            raise ValueError(f"nargs, if set, must be -1 not {nargs}")
         super(OptionEatAll, self).__init__(*args, **kwargs)
         self._previous_parser_process = None
         self._eat_all_parser = None

mysql_to_sqlite3/debug_info.py

@@ -52,9 +52,7 @@ def _implementation() -> str:
     else:
         implementation_version = "Unknown"
 
-    return "{implementation} {implementation_version}".format(
-        implementation=implementation, implementation_version=implementation_version
-    )
+    return f"{implementation} {implementation_version}"
 
 
 def _mysql_version() -> str:
@@ -73,10 +71,7 @@ def _mysql_version() -> str:
 def info() -> t.List[t.List[str]]:
     """Generate information for a bug report."""
     try:
-        platform_info: str = "{system} {release}".format(
-            system=platform.system(),
-            release=platform.release(),
-        )
+        platform_info: str = f"{platform.system()} {platform.release()}"
     except IOError:
         platform_info = "Unknown"
 

mysql_to_sqlite3/sqlite_utils.py

@@ -51,4 +51,4 @@ def convert_date(value: t.Any) -> date:
     try:
         return date.fromisoformat(value.decode())
     except ValueError as err:
-        raise ValueError("DATE field contains {}".format(err))  # pylint: disable=W0707
+        raise ValueError(f"DATE field contains {err}")  # pylint: disable=W0707

mysql_to_sqlite3/transporter.py

@@ -261,7 +261,7 @@ def _translate_default_from_mysql_to_sqlite(
                 "TINYBLOB",
                 "VARBINARY",
             }:
-                return "DEFAULT x'{}'".format(column_default.hex())
+                return f"DEFAULT x'{column_default.hex()}'"
             try:
                 column_default = column_default.decode()
             except (UnicodeDecodeError, AttributeError):
@@ -273,15 +273,15 @@ def _translate_default_from_mysql_to_sqlite(
                 if column_default:
                     return "DEFAULT(TRUE)"
                 return "DEFAULT(FALSE)"
-            return "DEFAULT '{}'".format(int(column_default))
+            return f"DEFAULT '{int(column_default)}'"
         if isinstance(column_default, str):
             if column_default.upper() in {
                 "CURRENT_TIME",
                 "CURRENT_DATE",
                 "CURRENT_TIMESTAMP",
             }:
-                return "DEFAULT {}".format(column_default.upper())
-        return "DEFAULT '{}'".format(str(column_default))
+                return f"DEFAULT {column_default.upper()}"
+        return f"DEFAULT '{str(column_default)}'"
 
     @classmethod
     def _data_type_collation_sequence(
@@ -297,7 +297,7 @@ def _data_type_collation_sequence(
                     "VARCHAR",
                 )
             ):
-                return "COLLATE {collation}".format(collation=collation)
+                return f"COLLATE {collation}"
         return ""
 
     def _check_sqlite_json1_extension_enabled(self) -> bool:
@@ -308,11 +308,11 @@ def _check_sqlite_json1_extension_enabled(self) -> bool:
             return False
 
     def _build_create_table_sql(self, table_name: str) -> str:
-        sql: str = 'CREATE TABLE IF NOT EXISTS "{}" ('.format(table_name)
+        sql: str = f'CREATE TABLE IF NOT EXISTS "{table_name}" ('
         primary: str = ""
         indices: str = ""
 
-        self._mysql_cur_dict.execute("SHOW COLUMNS FROM `{}`".format(table_name))
+        self._mysql_cur_dict.execute(f"SHOW COLUMNS FROM `{table_name}`")
 
         for row in self._mysql_cur_dict.fetchall():
             if row is not None:
@@ -351,8 +351,8 @@ def _build_create_table_sql(self, table_name: str) -> str:
 
                 if len(columns) > 0:
                     if index["primary"] in {1, "1"}:
-                        primary += "\n\tPRIMARY KEY ({columns})".format(
-                            columns=", ".join('"{}"'.format(column) for column in columns.split(","))
+                        primary += "\n\tPRIMARY KEY ({})".format(
+                            ", ".join(f'"{column}"' for column in columns.split(","))
                         )
                     else:
                         indices += """CREATE {unique} INDEX IF NOT EXISTS "{name}" ON "{table}" ({columns});""".format(
@@ -366,7 +366,7 @@ def _build_create_table_sql(self, table_name: str) -> str:
                             if isinstance(index["name"], bytes)
                             else index["name"],
                             table=table_name,
-                            columns=", ".join('"{}"'.format(column) for column in columns.split(",")),
+                            columns=", ".join(f'"{column}"' for column in columns.split(",")),
                         )
 
         sql += primary
@@ -554,18 +554,12 @@ def transfer(self) -> None:
                     if self._limit_rows > 0:
                         # limit to the requested number of rows
                         self._mysql_cur_dict.execute(
-                            """
-                                SELECT COUNT(*) AS `total_records`
-                                FROM (SELECT * FROM `{table_name}` LIMIT {limit}) AS `table`
-                            """.format(
-                                table_name=table_name, limit=self._limit_rows
-                            )
+                            "SELECT COUNT(*) AS `total_records` "
+                            f"FROM (SELECT * FROM `{table_name}` LIMIT {self._limit_rows}) AS `table`"
                         )
                     else:
                         # get all rows
-                        self._mysql_cur_dict.execute(
-                            "SELECT COUNT(*) AS `total_records` FROM `{table_name}`".format(table_name=table_name)
-                        )
+                        self._mysql_cur_dict.execute(f"SELECT COUNT(*) AS `total_records` FROM `{table_name}`")
 
                     total_records: t.Optional[t.Dict[str, ToPythonOutputTypes]] = self._mysql_cur_dict.fetchone()
                     if total_records is not None:
@@ -579,7 +573,7 @@ def transfer(self) -> None:
                         self._mysql_cur.execute(
                             "SELECT * FROM `{table_name}` {limit}".format(
                                 table_name=table_name,
-                                limit="LIMIT {}".format(self._limit_rows) if self._limit_rows > 0 else "",
+                                limit=f"LIMIT {self._limit_rows}" if self._limit_rows > 0 else "",
                             )
                         )
                         columns: t.Tuple[str, ...] = tuple(column[0] for column in self._mysql_cur.description)  # type: ignore[union-attr]

tests/conftest.py

@@ -117,7 +117,7 @@ def not_raises(exception: t.Type[Exception]) -> t.Generator:
         try:
             yield
         except exception:
-            raise pytest.fail("DID RAISE {0}".format(exception))
+            raise pytest.fail(f"DID RAISE {exception}")
 
     @staticmethod
     @contextmanager
@@ -142,7 +142,7 @@ def helpers() -> t.Type[Helpers]:
 @pytest.fixture()
 def sqlite_database(tmpdir: LocalPath) -> t.Union[str, Path, "os.PathLike[t.Any]"]:
     db_name: str = "".join(choice(ascii_uppercase + ascii_lowercase + digits) for _ in range(32))
-    return Path(tmpdir.join(Path("{}.sqlite3".format(db_name))))
+    return Path(tmpdir.join(Path(f"{db_name}.sqlite3")))
 
 
 def is_port_in_use(port: int, host: str = "0.0.0.0") -> bool:
@@ -178,9 +178,7 @@ def mysql_credentials(pytestconfig: Config) -> MySQLCredentials:
     if pytestconfig.getoption("use_docker"):
         while is_port_in_use(port, pytestconfig.getoption("mysql_host")):
             if port >= 2**16 - 1:
-                pytest.fail(
-                    "No ports appear to be available on the host {}".format(pytestconfig.getoption("mysql_host"))
-                )
+                pytest.fail(f"No ports appear to be available on the host {pytestconfig.getoption('mysql_host')}")
             port += 1
 
     return MySQLCredentials(
@@ -217,7 +215,7 @@ def mysql_instance(mysql_credentials: MySQLCredentials, pytestconfig: Config) ->
         docker_mysql_image = pytestconfig.getoption("docker_mysql_image") or "mysql:latest"
 
         if not any(docker_mysql_image in image.tags for image in client.images.list()):
-            print("Attempting to download Docker image {}'".format(docker_mysql_image))
+            print(f"Attempting to download Docker image {docker_mysql_image}'")
             try:
                 client.images.pull(docker_mysql_image)
             except (HTTPError, NotFound) as err:
@@ -226,12 +224,7 @@ def mysql_instance(mysql_credentials: MySQLCredentials, pytestconfig: Config) ->
         container = client.containers.run(
             image=docker_mysql_image,
             name="pytest_mysql_to_sqlite3",
-            ports={
-                "3306/tcp": (
-                    mysql_credentials.host,
-                    "{}/tcp".format(mysql_credentials.port),
-                )
-            },
+            ports={"3306/tcp": (mysql_credentials.host, f"{mysql_credentials.port}/tcp")},
             environment={
                 "MYSQL_RANDOM_ROOT_PASSWORD": "yes",
                 "MYSQL_USER": mysql_credentials.user,
@@ -285,13 +278,7 @@ def mysql_database(
     temp_image_dir: LocalPath = tmpdir_factory.mktemp("images")
 
     db: database.Database = database.Database(
-        "mysql+mysqldb://{user}:{password}@{host}:{port}/{database}".format(
-            user=mysql_credentials.user,
-            password=mysql_credentials.password,
-            host=mysql_credentials.host,
-            port=mysql_credentials.port,
-            database=mysql_credentials.database,
-        )
+        f"mysql+mysqldb://{mysql_credentials.user}:{mysql_credentials.password}@{mysql_credentials.host}:{mysql_credentials.port}/{mysql_credentials.database}"
     )
 
     with Helpers.session_scope(db) as session:

tests/func/mysql_to_sqlite3_test.py

@@ -417,7 +417,7 @@ def test_transfer_transfers_all_tables_from_mysql_to_sqlite(
         assert not any(record.levelname == "ERROR" for record in caplog.records)
 
         sqlite_engine: Engine = create_engine(
-            "sqlite:///{database}".format(database=sqlite_database),
+            f"sqlite:///{sqlite_database}",
             json_serializer=json.dumps,
             json_deserializer=json.loads,
         )
@@ -426,13 +426,7 @@ def test_transfer_transfers_all_tables_from_mysql_to_sqlite(
         sqlite_inspect: Inspector = inspect(sqlite_engine)
         sqlite_tables: t.List[str] = sqlite_inspect.get_table_names()
         mysql_engine: Engine = create_engine(
-            "mysql+mysqldb://{user}:{password}@{host}:{port}/{database}".format(
-                user=mysql_credentials.user,
-                password=mysql_credentials.password,
-                host=mysql_credentials.host,
-                port=mysql_credentials.port,
-                database=mysql_credentials.database,
-            )
+            f"mysql+mysqldb://{mysql_credentials.user}:{mysql_credentials.password}@{mysql_credentials.host}:{mysql_credentials.port}/{mysql_credentials.database}"
         )
         mysql_cnx: Connection = mysql_engine.connect()
         mysql_inspect: Inspector = inspect(mysql_engine)
@@ -466,10 +460,7 @@ def test_transfer_transfers_all_tables_from_mysql_to_sqlite(
                 mysql_index: t.Dict[str, t.Any] = {}
                 for key in index_keys:
                     if key == "name" and prefix_indices:
-                        mysql_index[key] = "{table}_{name}".format(
-                            table=table_name,
-                            name=index[key],  # type: ignore[literal-required]
-                        )
+                        mysql_index[key] = f"{table_name}_{index[key]}"  # type: ignore[literal-required]
                     else:
                         mysql_index[key] = index[key]  # type: ignore[literal-required]
                 mysql_indices.append(t.cast(ReflectedIndex, mysql_index))
@@ -510,7 +501,7 @@ def test_transfer_transfers_all_tables_from_mysql_to_sqlite(
             mysql_fk_result: CursorResult = mysql_cnx.execute(mysql_fk_stmt)
             mysql_foreign_keys: t.List[t.Dict[str, t.Any]] = [dict(row) for row in mysql_fk_result.mappings()]
 
-            sqlite_fk_stmt: TextClause = text('PRAGMA foreign_key_list("{table}")'.format(table=table_name))
+            sqlite_fk_stmt: TextClause = text(f'PRAGMA foreign_key_list("{table_name}")')
             sqlite_fk_result: CursorResult = sqlite_cnx.execute(sqlite_fk_stmt)
             if sqlite_fk_result.returns_rows:
                 for row in sqlite_fk_result.mappings():
@@ -884,13 +875,7 @@ def test_transfer_specific_tables_transfers_only_specified_tables_from_mysql_to_
         exclude_tables: bool,
     ) -> None:
         mysql_engine: Engine = create_engine(
-            "mysql+mysqldb://{user}:{password}@{host}:{port}/{database}".format(
-                user=mysql_credentials.user,
-                password=mysql_credentials.password,
-                host=mysql_credentials.host,
-                port=mysql_credentials.port,
-                database=mysql_credentials.database,
-            )
+            f"mysql+mysqldb://{mysql_credentials.user}:{mysql_credentials.password}@{mysql_credentials.host}:{mysql_credentials.port}/{mysql_credentials.database}"
         )
         mysql_cnx: Connection = mysql_engine.connect()
         mysql_inspect: Inspector = inspect(mysql_engine)
@@ -921,7 +906,7 @@ def test_transfer_specific_tables_transfers_only_specified_tables_from_mysql_to_
             message in [record.message for record in caplog.records]
             for message in set(
                 [
-                    "Transferring table {}".format(table)
+                    f"Transferring table {table}"
                     for table in (remaining_tables if exclude_tables else random_mysql_tables)
                 ]
                 + ["Done!"]
@@ -931,7 +916,7 @@ def test_transfer_specific_tables_transfers_only_specified_tables_from_mysql_to_
         assert not any(record.levelname == "ERROR" for record in caplog.records)
 
         sqlite_engine: Engine = create_engine(
-            "sqlite:///{database}".format(database=sqlite_database),
+            f"sqlite:///{sqlite_database}",
             json_serializer=json.dumps,
             json_deserializer=json.loads,
         )
@@ -960,10 +945,7 @@ def test_transfer_specific_tables_transfers_only_specified_tables_from_mysql_to_
                 mysql_index: t.Dict[str, t.Any] = {}
                 for key in index_keys:
                     if key == "name" and prefix_indices:
-                        mysql_index[key] = "{table}_{name}".format(
-                            table=table_name,
-                            name=index[key],  # type: ignore[literal-required]
-                        )
+                        mysql_index[key] = f"{table_name}_{index[key]}"  # type: ignore[literal-required]
                     else:
                         mysql_index[key] = index[key]  # type: ignore[literal-required]
                 mysql_indices.append(t.cast(ReflectedIndex, mysql_index))
@@ -1185,7 +1167,7 @@ def test_transfer_limited_rows_from_mysql_to_sqlite(
         assert not any(record.levelname == "ERROR" for record in caplog.records)
 
         sqlite_engine: Engine = create_engine(
-            "sqlite:///{database}".format(database=sqlite_database),
+            f"sqlite:///{sqlite_database}",
             json_serializer=json.dumps,
             json_deserializer=json.loads,
         )
@@ -1194,13 +1176,7 @@ def test_transfer_limited_rows_from_mysql_to_sqlite(
         sqlite_inspect: Inspector = inspect(sqlite_engine)
         sqlite_tables: t.List[str] = sqlite_inspect.get_table_names()
         mysql_engine: Engine = create_engine(
-            "mysql+mysqldb://{user}:{password}@{host}:{port}/{database}".format(
-                user=mysql_credentials.user,
-                password=mysql_credentials.password,
-                host=mysql_credentials.host,
-                port=mysql_credentials.port,
-                database=mysql_credentials.database,
-            )
+            f"mysql+mysqldb://{mysql_credentials.user}:{mysql_credentials.password}@{mysql_credentials.host}:{mysql_credentials.port}/{mysql_credentials.database}"
         )
         mysql_cnx: Connection = mysql_engine.connect()
         mysql_inspect: Inspector = inspect(mysql_engine)
@@ -1234,10 +1210,7 @@ def test_transfer_limited_rows_from_mysql_to_sqlite(
                 mysql_index: t.Dict[str, t.Any] = {}
                 for key in index_keys:
                     if key == "name" and prefix_indices:
-                        mysql_index[key] = "{table}_{name}".format(
-                            table=table_name,
-                            name=index[key],  # type: ignore[literal-required]
-                        )
+                        mysql_index[key] = f"{table_name}_{index[key]}"  # type: ignore[literal-required]
                     else:
                         mysql_index[key] = index[key]  # type: ignore[literal-required]
                 mysql_indices.append(t.cast(ReflectedIndex, mysql_index))
@@ -1278,7 +1251,7 @@ def test_transfer_limited_rows_from_mysql_to_sqlite(
             mysql_fk_result: CursorResult = mysql_cnx.execute(mysql_fk_stmt)
             mysql_foreign_keys: t.List[t.Dict[str, t.Any]] = [dict(row) for row in mysql_fk_result.mappings()]
 
-            sqlite_fk_stmt: TextClause = text('PRAGMA foreign_key_list("{table}")'.format(table=table_name))
+            sqlite_fk_stmt: TextClause = text(f'PRAGMA foreign_key_list("{table_name}")')
             sqlite_fk_result: CursorResult = sqlite_cnx.execute(sqlite_fk_stmt)
             if sqlite_fk_result.returns_rows:
                 for row in sqlite_fk_result.mappings():

tests/func/test_cli.py

@@ -374,13 +374,7 @@ def test_specific_tables_include_and_exclude_are_mutually_exclusive_options(
         mysql_database: Database,
     ) -> None:
         mysql_engine: Engine = create_engine(
-            "mysql+mysqldb://{user}:{password}@{host}:{port}/{database}".format(
-                user=mysql_credentials.user,
-                password=mysql_credentials.password,
-                host=mysql_credentials.host,
-                port=mysql_credentials.port,
-                database=mysql_credentials.database,
-            )
+            f"mysql+mysqldb://{mysql_credentials.user}:{mysql_credentials.password}@{mysql_credentials.host}:{mysql_credentials.port}/{mysql_credentials.database}"
         )
         mysql_cnx: Connection = mysql_engine.connect()
         mysql_inspect: Inspector = inspect(mysql_engine)
@@ -428,13 +422,7 @@ def test_transfer_specific_tables_only(
         mysql_database: Database,
     ) -> None:
         mysql_engine: Engine = create_engine(
-            "mysql+mysqldb://{user}:{password}@{host}:{port}/{database}".format(
-                user=mysql_credentials.user,
-                password=mysql_credentials.password,
-                host=mysql_credentials.host,
-                port=mysql_credentials.port,
-                database=mysql_credentials.database,
-            )
+            f"mysql+mysqldb://{mysql_credentials.user}:{mysql_credentials.password}@{mysql_credentials.host}:{mysql_credentials.port}/{mysql_credentials.database}"
         )
         mysql_inspect: Inspector = inspect(mysql_engine)
         mysql_tables: t.List[str] = mysql_inspect.get_table_names()