⚡ use f-strings where appropriate

Author: techouse

sqlite3_to_mysql/cli.py

@@ -147,12 +147,7 @@ def cli(
             )
             if mysql_collation not in set(charset_collations):
                 raise click.ClickException(
-                    "Error: Invalid value for '--collation' of charset '{charset}': '{collation}' is not one of "
-                    "{collations}.".format(
-                        collation=mysql_collation,
-                        charset=mysql_charset,
-                        collations="'" + "', '".join(charset_collations) + "'",
-                    )
+                    f"""Error: Invalid value for '--collation' of charset '{mysql_charset}': '{mysql_collation}' is not one of {"'" + "', '".join(charset_collations) + "'"}."""
                 )
 
         SQLite3toMySQL(

sqlite3_to_mysql/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

sqlite3_to_mysql/debug_info.py

@@ -51,9 +51,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:
@@ -72,10 +70,7 @@ def _mysql_version() -> str:
 def info() -> t.List[t.List[str]]:
     """Generate information for a bug report."""
     try:
-        platform_info = "{system} {release}".format(
-            system=platform.system(),
-            release=platform.release(),
-        )
+        platform_info = f"{platform.system()} {platform.release()}"
     except IOError:
         platform_info = "Unknown"
 

sqlite3_to_mysql/sqlite_utils.py

@@ -43,7 +43,7 @@ def convert_date(value) -> 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
 
 
 def check_sqlite_table_xinfo_support(version_string: str) -> bool:

sqlite3_to_mysql/transporter.py

@@ -215,7 +215,7 @@ def _get_sqlite_version(self) -> str:
 
     def _sqlite_table_has_rowid(self, table: str) -> bool:
         try:
-            self._sqlite_cur.execute("""SELECT rowid FROM "{}" LIMIT 1""".format(table))
+            self._sqlite_cur.execute(f'SELECT rowid FROM "{table}" LIMIT 1')
             self._sqlite_cur.fetchall()
             return True
         except sqlite3.OperationalError:
@@ -224,15 +224,11 @@ def _sqlite_table_has_rowid(self, table: str) -> bool:
     def _create_database(self) -> None:
         try:
             self._mysql_cur.execute(
-                """
-                CREATE DATABASE IF NOT EXISTS `{database}`
-                DEFAULT CHARACTER SET {charset}
-                DEFAULT COLLATE {collation}
-            """.format(
-                    database=self._mysql_database,
-                    charset=self._mysql_charset,
-                    collation=self._mysql_collation,
-                )
+                f"""
+                CREATE DATABASE IF NOT EXISTS `{self._mysql_database}`
+                DEFAULT CHARACTER SET {self._mysql_charset}
+                DEFAULT COLLATE {self._mysql_collation}
+            """
             )
             self._mysql_cur.close()
             self._mysql.commit()
@@ -300,23 +296,21 @@ def _column_type_length(cls, column_type: str, default: t.Optional[t.Union[str,
         if suffix:
             return suffix.group(0)
         if default:
-            return "({})".format(default)
+            return f"({default})"
         return ""
 
     def _create_table(self, table_name: str, transfer_rowid: bool = False) -> None:
         primary_keys: t.List[t.Dict[str, str]] = []
 
-        sql: str = "CREATE TABLE IF NOT EXISTS `{}` ( ".format(safe_identifier_length(table_name))
+        sql: str = f"CREATE TABLE IF NOT EXISTS `{safe_identifier_length(table_name)}` ( "
 
         if transfer_rowid:
             sql += " `rowid` BIGINT NOT NULL, "
 
-        self._sqlite_cur.execute(
-            'PRAGMA {command}("{table_name}")'.format(
-                command="table_xinfo" if self._sqlite_table_xinfo_support else "table_info",
-                table_name=table_name,
-            )
-        )
+        if self._sqlite_table_xinfo_support:
+            self._sqlite_cur.execute(f'PRAGMA table_xinfo("{table_name}")')
+        else:
+            self._sqlite_cur.execute(f'PRAGMA table_info("{table_name}")')
 
         rows: t.List[t.Any] = self._sqlite_cur.fetchall()
         compound_primary_key: bool = len(tuple(True for row in rows if dict(row)["pk"] > 0)) > 1
@@ -364,12 +358,9 @@ def _create_table(self, table_name: str, transfer_rowid: bool = False) -> None:
             )
 
         if transfer_rowid:
-            sql += ", CONSTRAINT `{}_rowid` UNIQUE (`rowid`)".format(safe_identifier_length(table_name))
+            sql += f", CONSTRAINT `{safe_identifier_length(table_name)}_rowid` UNIQUE (`rowid`)"
 
-        sql += " ) ENGINE=InnoDB DEFAULT CHARSET={charset} COLLATE={collation}".format(
-            charset=self._mysql_charset,
-            collation=self._mysql_collation,
-        )
+        sql += f" ) ENGINE=InnoDB DEFAULT CHARSET={self._mysql_charset} COLLATE={self._mysql_collation}"
 
         try:
             self._mysql_cur.execute(sql)
@@ -395,27 +386,23 @@ def _truncate_table(self, table_name: str) -> None:
         )
         if len(self._mysql_cur.fetchall()) > 0:
             self._logger.info("Truncating table %s", safe_identifier_length(table_name))
-            self._mysql_cur.execute(
-                "TRUNCATE TABLE `{}`".format(
-                    safe_identifier_length(table_name),
-                ),
-            )
+            self._mysql_cur.execute(f"TRUNCATE TABLE `{safe_identifier_length(table_name)}`")
 
     def _add_indices(self, table_name: str) -> None:
-        self._sqlite_cur.execute('PRAGMA table_info("{}")'.format(table_name))
+        self._sqlite_cur.execute(f'PRAGMA table_info("{table_name}")')
         table_columns: t.Dict[str, str] = {}
         for row in self._sqlite_cur.fetchall():
             column: t.Dict[str, t.Any] = dict(row)
             table_columns[column["name"]] = column["type"]
 
-        self._sqlite_cur.execute('PRAGMA index_list("{}")'.format(table_name))
+        self._sqlite_cur.execute(f'PRAGMA index_list("{table_name}")')
         indices: t.Tuple[t.Dict[str, t.Any], ...] = tuple(dict(row) for row in self._sqlite_cur.fetchall())
 
         for index in indices:
             if index["origin"] == "pk":
                 continue
 
-            self._sqlite_cur.execute('PRAGMA index_info("{}")'.format(index["name"]))
+            self._sqlite_cur.execute(f'PRAGMA index_info("{index["name"]}")')
             index_infos: t.Tuple[t.Dict[str, t.Any], ...] = tuple(dict(row) for row in self._sqlite_cur.fetchall())
 
             index_type: str = "UNIQUE" if int(index["unique"]) == 1 else "INDEX"
@@ -428,14 +415,14 @@ def _add_indices(self, table_name: str) -> None:
                     # Use fulltext if requested and available
                     index_type = "FULLTEXT"
                     index_columns: str = ",".join(
-                        "`{}`".format(safe_identifier_length(index_info["name"])) for index_info in index_infos
+                        f'`{safe_identifier_length(index_info["name"])}`' for index_info in index_infos
                     )
                 else:
                     # Limit the max TEXT field index length to 255
                     index_columns = ", ".join(
                         "`{column}`{length}".format(
                             column=safe_identifier_length(index_info["name"]),
-                            length="({})".format(255)
+                            length="(255)"
                             if table_columns[index_info["name"]].upper() in MYSQL_TEXT_COLUMN_TYPES_WITH_JSON
                             else "",
                         )
@@ -447,19 +434,14 @@ def _add_indices(self, table_name: str) -> None:
                     index_length: str = ""
                     # Limit the max BLOB field index length to 255
                     if table_columns[index_info["name"]].upper() in MYSQL_BLOB_COLUMN_TYPES:
-                        index_length = "({})".format(255)
+                        index_length = "(255)"
                     else:
                         suffix: t.Optional[t.Match[str]] = self.COLUMN_LENGTH_PATTERN.search(
                             table_columns[index_info["name"]]
                         )
                         if suffix:
                             index_length = suffix.group(0)
-                    column_list.append(
-                        "`{column}`{length}".format(
-                            column=safe_identifier_length(index_info["name"]),
-                            length=index_length,
-                        )
-                    )
+                    column_list.append(f'`{safe_identifier_length(index_info["name"])}`{index_length}')
                 index_columns = ", ".join(column_list)
 
             try:
@@ -480,7 +462,7 @@ def _add_indices(self, table_name: str) -> None:
                         index_columns=", ".join(
                             "`{column}`{length}".format(
                                 column=safe_identifier_length(index_info["name"]),
-                                length="({})".format(255)
+                                length="(255)"
                                 if table_columns[index_info["name"]].upper() in MYSQL_TEXT_COLUMN_TYPES_WITH_JSON
                                 else "",
                             )
@@ -508,10 +490,7 @@ def _add_index(
             index_type=index_type,
             name=safe_identifier_length(index["name"])
             if index_iteration == 0
-            else "{name}_{index_iteration}".format(
-                name=safe_identifier_length(index["name"], max_length=60),
-                index_iteration=index_iteration,
-            ),
+            else f'{safe_identifier_length(index["name"], max_length=60)}_{index_iteration}',
             columns=index_columns,
         )
 
@@ -567,7 +546,7 @@ def _add_index(
                 raise
 
     def _add_foreign_keys(self, table_name: str) -> None:
-        self._sqlite_cur.execute('PRAGMA foreign_key_list("{}")'.format(table_name))
+        self._sqlite_cur.execute(f'PRAGMA foreign_key_list("{table_name}")')
 
         for row in self._sqlite_cur.fetchall():
             foreign_key: t.Dict[str, t.Any] = dict(row)
@@ -640,14 +619,12 @@ def transfer(self) -> None:
         if len(self._sqlite_tables) > 0:
             # transfer only specific tables
             self._sqlite_cur.execute(
-                """
+                f"""
                 SELECT name FROM sqlite_master
                 WHERE type='table'
                 AND name NOT LIKE 'sqlite_%'
-                AND name IN({placeholders})
-                """.format(
-                    placeholders=("?, " * len(self._sqlite_tables)).rstrip(" ,")
-                ),
+                AND name IN({("?, " * len(self._sqlite_tables)).rstrip(" ,")})
+                """,
                 self._sqlite_tables,
             )
         else:
@@ -676,17 +653,15 @@ def transfer(self) -> None:
                     self._truncate_table(table["name"])
 
                 # get the size of the data
-                self._sqlite_cur.execute('SELECT COUNT(*) AS total_records FROM "{}"'.format(table["name"]))
+                self._sqlite_cur.execute(f'SELECT COUNT(*) AS total_records FROM "{table["name"]}"')
                 total_records = int(dict(self._sqlite_cur.fetchone())["total_records"])
 
                 # only continue if there is anything to transfer
                 if total_records > 0:
                     # populate it
                     self._logger.info("Transferring table %s", table["name"])
                     self._sqlite_cur.execute(
-                        """
-                        SELECT {rowid} * FROM "{table_name}"
-                    """.format(
+                        '''SELECT {rowid} * FROM "{table_name}"'''.format(
                             rowid='rowid as "rowid",' if transfer_rowid else "",
                             table_name=table["name"],
                         )

tests/conftest.py

@@ -124,7 +124,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
@@ -151,13 +151,13 @@ def sqlite_database(pytestconfig: Config, _session_faker: Faker, tmpdir_factory:
     db_file: LocalPath = pytestconfig.getoption("sqlite_file")
     if db_file:
         if not isfile(realpath(db_file)):
-            pytest.fail("{} does not exist".format(db_file))
+            pytest.fail(f"{db_file} does not exist")
         return str(realpath(db_file))
 
     temp_data_dir: LocalPath = tmpdir_factory.mktemp("data")
     temp_image_dir: LocalPath = tmpdir_factory.mktemp("images")
     db_file = temp_data_dir.join(Path("db.sqlite3"))
-    db: Database = Database("sqlite:///{}".format(str(db_file)))
+    db: Database = Database(f"sqlite:///{db_file}")
 
     with Helpers.session_scope(db) as session:
         for _ in range(_session_faker.pyint(min_value=12, max_value=24)):
@@ -220,9 +220,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(
@@ -259,7 +257,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:
@@ -271,7 +269,7 @@ def mysql_instance(mysql_credentials: MySQLCredentials, pytestconfig: Config) ->
             ports={
                 "3306/tcp": (
                     mysql_credentials.host,
-                    "{}/tcp".format(mysql_credentials.port),
+                    f"{mysql_credentials.port}/tcp",
                 )
             },
             environment={
@@ -322,13 +320,7 @@ def mysql_database(mysql_instance: t.Generator, mysql_credentials: MySQLCredenti
     yield  # type: ignore[misc]
 
     engine: Engine = create_engine(
-        "mysql+pymysql://{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+pymysql://{mysql_credentials.user}:{mysql_credentials.password}@{mysql_credentials.host}:{mysql_credentials.port}/{mysql_credentials.database}"
     )
 
     if database_exists(engine.url):