Upgrade f-strings with flynt -a and remove int()

lib/extras.py

@@ -477,7 +477,7 @@ class MinTimeLoggingConnection(LoggingConnection):
         if t > self._mintime:
             if isinstance(msg, bytes):
                 msg = msg.decode(_ext.encodings[self.encoding], 'replace')
-            return msg + _os.linesep + "  (execution time: %d ms)" % t
+            return f"{msg}{_os.linesep}  (execution time: {t} ms)"
 
     def cursor(self, *args, **kwargs):
         kwargs.setdefault('cursor_factory',
@@ -603,7 +603,7 @@ class ReplicationCursor(_replicationCursor):
                 raise psycopg2.ProgrammingError(
                     "cannot specify timeline for logical replication")
 
-            command += " TIMELINE %d" % timeline
+            command += f" TIMELINE {timeline}"
 
         if options:
             if slot_type == REPLICATION_PHYSICAL:
@@ -869,7 +869,7 @@ class HstoreAdapter:
         for m in self._re_hstore.finditer(s):
             if m is None or m.start() != start:
                 raise psycopg2.InterfaceError(
-                    "error parsing hstore pair at char %d" % start)
+                    f"error parsing hstore pair at char {start}")
             k = _bsdec.sub(r'\1', m.group(1))
             v = m.group(2)
             if v is not None:
@@ -880,7 +880,7 @@ class HstoreAdapter:
 
         if start < len(s):
             raise psycopg2.InterfaceError(
-                "error parsing hstore: unparsed data after char %d" % start)
+                f"error parsing hstore: unparsed data after char {start}")
 
         return rv
 

scripts/refcounter.py

@@ -39,24 +39,24 @@ def main():
     sys.stdout.write(f"test suite {test.__name__}\n")
 
     for i in range(1, opt.nruns + 1):
-        sys.stdout.write("test suite run %d of %d\n" % (i, opt.nruns))
+        sys.stdout.write(f"test suite run {i} of {opt.nruns}\n")
         runner = unittest.TextTestRunner()
         runner.run(test.test_suite())
         dump(i, opt)
 
-    f1 = open('debug-%02d.txt' % (opt.nruns - 1)).readlines()
+    f1 = open(f'debug-{(opt.nruns - 1):02}.txt').readlines()
-    f2 = open('debug-%02d.txt' % opt.nruns).readlines()
+    f2 = open(f'debug-{opt.nruns:02}.txt').readlines()
     for line in difflib.unified_diff(f1, f2,
-            "run %d" % (opt.nruns - 1), "run %d" % opt.nruns):
+            f"run {opt.nruns - 1}", f"run {opt.nruns}"):
         sys.stdout.write(line)
 
     rv = f1 != f2 and 1 or 0
 
     if opt.objs:
-        f1 = open('objs-%02d.txt' % (opt.nruns - 1)).readlines()
+        f1 = open(f'objs-{(opt.nruns - 1):02}.txt').readlines()
-        f2 = open('objs-%02d.txt' % opt.nruns).readlines()
+        f2 = open(f'objs-{opt.nruns:02}.txt').readlines()
         for line in difflib.unified_diff(f1, f2,
-                "run %d" % (opt.nruns - 1), "run %d" % opt.nruns):
+                f"run {opt.nruns - 1}", f"run {opt.nruns}"):
             sys.stdout.write(line)
 
     return rv
@@ -85,7 +85,7 @@ def dump(i, opt):
 
     pprint(
         sorted(((v, str(k)) for k, v in c.items()), reverse=True),
-        stream=open("debug-%02d.txt" % i, "w"))
+        stream=open(f"debug-{i:02}.txt", "w"))
 
     if opt.objs:
         co = []
@@ -100,7 +100,7 @@ def dump(i, opt):
         else:
             co.sort()
 
-        pprint(co, stream=open("objs-%02d.txt" % i, "w"))
+        pprint(co, stream=open(f"objs-{i:02}.txt", "w"))
 
 
 if __name__ == '__main__':

tests/dbapi20_tpc.py

@@ -19,7 +19,7 @@ class TwoPhaseCommitTests(unittest.TestCase):
     def make_xid(self, con):
         id = TwoPhaseCommitTests._last_id
         TwoPhaseCommitTests._last_id += 1
-        return con.xid(42, "%s%d" % (self._global_id_prefix, id), "qualifier")
+        return con.xid(42, f"{self._global_id_prefix}{id}", "qualifier")
 
     def test_xid(self):
         con = self.connect()

tests/test_connection.py

@@ -302,7 +302,7 @@ class ConnectionTests(ConnectingTestCase):
         # Stop the committer thread
         stop.append(True)
 
-        self.assert_(not notices, "%d notices raised" % len(notices))
+        self.assert_(not notices, f"{len(notices)} notices raised")
 
     def test_connect_cursor_factory(self):
         conn = self.connect(cursor_factory=psycopg2.extras.DictCursor)

tests/test_copy.py

@@ -246,7 +246,7 @@ class CopyTests(ConnectingTestCase):
             curs.copy_expert, 'COPY tcopy (data) FROM STDIN', f)
 
     def test_copy_no_column_limit(self):
-        cols = ["c%050d" % i for i in range(200)]
+        cols = [f"c{i:050}" for i in range(200)]
 
         curs = self.conn.cursor()
         curs.execute('CREATE TEMPORARY TABLE manycols (%s)' % ',\n'.join(

tests/testutils.py

@@ -309,7 +309,7 @@ def skip_before_libpq(*ver):
         v = libpq_version()
         decorator = unittest.skipIf(
             v < int("%d%02d%02d" % ver),
-            "skipped because libpq %d" % v,
+            f"skipped because libpq {v}",
         )
         return decorator(cls)
     return skip_before_libpq_