PyMySQL · methane · May 24, 2023 · May 24, 2023 · May 24, 2023 · May 24, 2023
diff --git a/pymysql/charset.py b/pymysql/charset.py
@@ -7,7 +7,7 @@ def __init__(self, id, name, collation, is_default):
         self.is_default = is_default == "Yes"
 
     def __repr__(self):
-        return "Charset(id=%s, name=%r, collation=%r)" % (
+        return "Charset(id={}, name={!r}, collation={!r})".format(
             self.id,
             self.name,
             self.collation,

diff --git a/pymysql/connections.py b/pymysql/connections.py
@@ -528,7 +528,9 @@ def escape_string(self, s):
 
     def _quote_bytes(self, s):
         if self.server_status & SERVER_STATUS.SERVER_STATUS_NO_BACKSLASH_ESCAPES:
-            return "'%s'" % (s.replace(b"'", b"''").decode("ascii", "surrogateescape"),)
+            return "'{}'".format(
+                s.replace(b"'", b"''").decode("ascii", "surrogateescape")
+            )
         return converters.escape_bytes(s)
 
     def cursor(self, cursor=None):
@@ -621,7 +623,7 @@ def connect(self, sock=None):
                                 (self.host, self.port), self.connect_timeout, **kwargs
                             )
                             break
-                        except (OSError, IOError) as e:
+                        except OSError as e:
                             if e.errno == errno.EINTR:
                                 continue
                             raise
@@ -662,7 +664,7 @@ def connect(self, sock=None):
             if isinstance(e, (OSError, IOError)):
                 exc = err.OperationalError(
                     CR.CR_CONN_HOST_ERROR,
-                    "Can't connect to MySQL server on %r (%s)" % (self.host, e),
+                    f"Can't connect to MySQL server on {self.host!r} ({e})",
                 )
                 # Keep original exception and traceback to investigate error.
                 exc.original_exception = e
@@ -739,13 +741,13 @@ def _read_bytes(self, num_bytes):
             try:
                 data = self._rfile.read(num_bytes)
                 break
-            except (IOError, OSError) as e:
+            except OSError as e:
                 if e.errno == errno.EINTR:
                     continue
                 self._force_close()
                 raise err.OperationalError(
                     CR.CR_SERVER_LOST,
-                    "Lost connection to MySQL server during query (%s)" % (e,),
+                    f"Lost connection to MySQL server during query ({e})",
                 )
             except BaseException:
                 # Don't convert unknown exception to MySQLError.
@@ -762,10 +764,10 @@ def _write_bytes(self, data):
         self._sock.settimeout(self._write_timeout)
         try:
             self._sock.sendall(data)
-        except IOError as e:
+        except OSError as e:
             self._force_close()
             raise err.OperationalError(
-                CR.CR_SERVER_GONE_ERROR, "MySQL server has gone away (%r)" % (e,)
+                CR.CR_SERVER_GONE_ERROR, f"MySQL server has gone away ({e!r})"
             )
 
     def _read_query_result(self, unbuffered=False):
@@ -1006,7 +1008,7 @@ def _process_auth(self, plugin_name, auth_packet):
                 else:
                     raise err.OperationalError(
                         CR.CR_AUTH_PLUGIN_CANNOT_LOAD,
-                        "Authentication plugin '%s' not configured" % (plugin_name,),
+                        f"Authentication plugin '{plugin_name}' not configured",
                     )
                 pkt = self._read_packet()
                 pkt.check_error()
@@ -1382,7 +1384,7 @@ def send_data(self):
                     if not chunk:
                         break
                     conn.write_packet(chunk)
-        except IOError:
+        except OSError:
             raise err.OperationalError(
                 ER.FILE_NOT_FOUND,
                 f"Can't find file '{self.filename}'",

diff --git a/pymysql/cursors.py b/pymysql/cursors.py
@@ -262,7 +262,7 @@ def callproc(self, procname, args=()):
             )
             self.nextset()
 
-        q = "CALL %s(%s)" % (
+        q = "CALL {}({})".format(
             procname,
             ",".join(["@_%s_%d" % (procname, i) for i in range(len(args))]),
         )
@@ -383,7 +383,7 @@ class DictCursorMixin:
     dict_type = dict
 
     def _do_get_result(self):
-        super(DictCursorMixin, self)._do_get_result()
+        super()._do_get_result()
         fields = []
         if self.description:
             for f in self._result.fields:

diff --git a/pymysql/protocol.py b/pymysql/protocol.py
@@ -35,7 +35,7 @@ def printable(data):
     dump_data = [data[i : i + 16] for i in range(0, min(len(data), 256), 16)]
     for d in dump_data:
         print(
-            " ".join("{:02X}".format(x) for x in d)
+            " ".join(f"{x:02X}" for x in d)
             + "   " * (16 - len(d))
             + " " * 2
             + "".join(printable(x) for x in d)
@@ -275,7 +275,7 @@ def get_column_length(self):
         return self.length
 
     def __str__(self):
-        return "%s %r.%r.%r, type=%s, flags=%x" % (
+        return "{} {!r}.{!r}.{!r}, type={}, flags={:x}".format(
             self.__class__,
             self.db,
             self.table_name,

diff --git a/pymysql/tests/base.py b/pymysql/tests/base.py
@@ -98,7 +98,7 @@ def safe_create_table(self, connection, tablename, ddl, cleanup=True):
 
         with warnings.catch_warnings():
             warnings.simplefilter("ignore")
-            cursor.execute("drop table if exists `%s`" % (tablename,))
+            cursor.execute(f"drop table if exists `{tablename}`")
         cursor.execute(ddl)
         cursor.close()
         if cleanup:
@@ -108,5 +108,5 @@ def drop_table(self, connection, tablename):
         cursor = connection.cursor()
         with warnings.catch_warnings():
             warnings.simplefilter("ignore")
-            cursor.execute("drop table if exists `%s`" % (tablename,))
+            cursor.execute(f"drop table if exists `{tablename}`")
         cursor.close()
diff --git a/pymysql/tests/test_DictCursor.py b/pymysql/tests/test_DictCursor.py
@@ -13,7 +13,7 @@ class TestDictCursor(base.PyMySQLTestCase):
     cursor_type = pymysql.cursors.DictCursor
 
     def setUp(self):
-        super(TestDictCursor, self).setUp()
+        super().setUp()
         self.conn = conn = self.connect()
         c = conn.cursor(self.cursor_type)
 
@@ -36,7 +36,7 @@ def setUp(self):
     def tearDown(self):
         c = self.conn.cursor()
         c.execute("drop table dictcursor")
-        super(TestDictCursor, self).tearDown()
+        super().tearDown()
 
     def _ensure_cursor_expired(self, cursor):
         pass

diff --git a/pymysql/tests/test_basic.py b/pymysql/tests/test_basic.py
@@ -332,7 +332,7 @@ class TestBulkInserts(base.PyMySQLTestCase):
     cursor_type = pymysql.cursors.DictCursor
 
     def setUp(self):
-        super(TestBulkInserts, self).setUp()
+        super().setUp()
         self.conn = conn = self.connect()
 
         # create a table and some data to query

diff --git a/pymysql/tests/test_connection.py b/pymysql/tests/test_connection.py
@@ -28,7 +28,7 @@ def __init__(self, c, user, db, auth=None, authdata=None, password=None):
             # already exists - TODO need to check the same plugin applies
             self._created = False
         try:
-            c.execute("GRANT SELECT ON %s.* TO %s" % (db, user))
+            c.execute(f"GRANT SELECT ON {db}.* TO {user}")
             self._grant = True
         except pymysql.err.InternalError:
             self._grant = False
@@ -38,7 +38,7 @@ def __enter__(self):
 
     def __exit__(self, exc_type, exc_value, traceback):
         if self._grant:
-            self._c.execute("REVOKE SELECT ON %s.* FROM %s" % (self._db, self._user))
+            self._c.execute(f"REVOKE SELECT ON {self._db}.* FROM {self._user}")
         if self._created:
             self._c.execute("DROP USER %s" % self._user)
 

diff --git a/pymysql/tests/test_cursor.py b/pymysql/tests/test_cursor.py
@@ -7,7 +7,7 @@
 
 class CursorTest(base.PyMySQLTestCase):
     def setUp(self):
-        super(CursorTest, self).setUp()
+        super().setUp()
 
         conn = self.connect()
         self.safe_create_table(

diff --git a/pymysql/tests/test_issues.py b/pymysql/tests/test_issues.py
@@ -379,8 +379,8 @@ def test_issue_175(self):
         conn = self.connect()
         cur = conn.cursor()
         for length in (200, 300):
-            columns = ", ".join("c{0} integer".format(i) for i in range(length))
-            sql = "create table test_field_count ({0})".format(columns)
+            columns = ", ".join(f"c{i} integer" for i in range(length))
+            sql = f"create table test_field_count ({columns})"
             try:
                 cur.execute(sql)
                 cur.execute("select * from test_field_count")