bpo-34828 sqlite3.iterdump now correctly handles tables with autoincr…

…ement The iterdump command in Lib/sqlite3/dump.py assumed that the table "sqlite_sequence" already exists. When this was not the case and exception was thrown.
python · erlend-aasland · Jun 19, 2022 · Sep 28, 2018 · Sep 28, 2018 · Jul 3, 2021
commit 9bf9eb7b5b2a9f174da7073826c9ca98d6c21b50
@@ -28,9 +28,14 @@ def _iterdump(connection):
             ORDER BY "name"
         """
     schema_res = cu.execute(q)
+    sqlite_sequence = []
     for table_name, type, sql in schema_res.fetchall():
         if table_name == 'sqlite_sequence':
-            yield('DELETE FROM "sqlite_sequence";')
+            sqlite_sequence = ['DELETE FROM "sqlite_sequence"']
+            rows = cu.execute('SELECT * FROM "sqlite_sequence";').fetchall()
+            sqlite_sequence.extend(['INSERT INTO "sqlite_sequence" VALUES(\'{0}\',{1})'
+                                   .format(row[0], row[1]) for row in rows])
+            continue
         elif table_name == 'sqlite_stat1':
             yield('ANALYZE "sqlite_master";')
         elif table_name.startswith('sqlite_'):
@@ -67,4 +72,6 @@ def _iterdump(connection):
     for name, type, sql in schema_res.fetchall():
         yield('{0};'.format(sql))
 
+    for row in sqlite_sequence:
+        yield("{0};".format(row))
     yield('COMMIT;')
diff --git a/Lib/sqlite3/test/dump.py b/Lib/sqlite3/test/dump.py
@@ -49,6 +49,67 @@ def CheckTableDump(self):
         [self.assertEqual(expected_sqls[i], actual_sqls[i])
             for i in range(len(expected_sqls))]
 
+    def CheckIterdumpAutoincrement(self):
+        expected_sqls = [
+            """CREATE TABLE "posts" (id int primary key);"""
+            ,
+            """INSERT INTO "posts" VALUES(0);"""
+            ,
+            "CREATE TABLE \"tags\" ( " \
+            "id integer primary key autoincrement," \
+            "tag varchar(256)," \
+            "post int references posts);"
+            ,
+            """INSERT INTO "tags" VALUES(NULL,'test',0);"""
+            ,
+            "CREATE TABLE \"tags2\" ( " \
+            "id integer primary key autoincrement," \
+            "tag varchar(256)," \
+            "post int references posts);"
+        ]
+        [self.cu.execute(s) for s in expected_sqls]
+        i = self.cx.iterdump()
+        actual_sqls = [s for s in i]
+        expected_sqls[3] = expected_sqls[3].replace("NULL", "1")
+        expected_sqls = ['BEGIN TRANSACTION;'] + expected_sqls + \
+                        ['DELETE FROM "sqlite_sequence";'] + \
+                        ["""INSERT INTO "sqlite_sequence" VALUES('tags',1);"""] + \
+                        ['COMMIT;']
+        [self.assertEqual(expected_sqls[i], actual_sqls[i])
+         for i in range(len(expected_sqls))]
+
+    def CheckIterdumpAutoincrementCreateNewDB(self):
+        old_db = [
+            "BEGIN TRANSACTION ;"
+            ,
+            """CREATE TABLE "posts" (id int primary key);"""
+            ,
+            """INSERT INTO "posts" VALUES(0);"""
+            ,
+            "CREATE TABLE \"tags\" ( " \
+            "id integer primary key autoincrement," \
+            "tag varchar(256)," \
+            "post int references posts);"
+            ,
+            "CREATE TABLE \"tags2\" ( " \
+            "id integer primary key autoincrement," \
+            "tag varchar(256)," \
+            "post int references posts);"
+        ]
+        [old_db.append("""INSERT INTO "tags" VALUES(NULL,'test{0}',0);""".format(i)) for i in range(1, 10)]
+        [old_db.append("""INSERT INTO "tags2" VALUES(NULL,'test{0}',0);""".format(i)) for i in range(1, 5)]
+        old_db.append("COMMIT;")
+        [self.cu.execute(s) for s in old_db]
+        i = self.cx.iterdump()
+        cx2 = sqlite.connect(":memory:")
+        query = "".join(line for line in self.cx.iterdump())
+        cx2.executescript(query)
+        cu2 = cx2.cursor()
+        self.assertEqual(cu2.execute('SELECT "seq" FROM "sqlite_sequence" WHERE "name" == "tags"').fetchall()[0][0], 9)
+        self.assertEqual(cu2.execute('SELECT "seq" FROM "sqlite_sequence" WHERE "name" == "tags2"').fetchall()[0][0], 4)
+        cu2.close()
+        cx2.close()
+
     def CheckUnorderableRow(self):
         # iterdump() should be able to cope with unorderable row types (issue #15545)
         class UnorderableRow: