gh-79009: sqlite3.iterdump now correctly handles tables with autoincrement

Lib/sqlite3/dump.py

@@ -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,8 @@ def _iterdump(connection):
     for name, type, sql in schema_res.fetchall():
         yield('{0};'.format(sql))
 
+    # Yield statements concerning the sqlite_sequence table at the end of the transaction: (bpo-34828)
+    for row in sqlite_sequence:
+        yield("{0};".format(row))
+
     yield('COMMIT;')

Lib/sqlite3/test/dump.py

@@ -1,7 +1,8 @@
 # Author: Paul Kippes <[email protected]>
 
-import unittest
 import sqlite3 as sqlite
+import unittest
+
 
 class DumpTests(unittest.TestCase):
     def setUp(self):
@@ -49,6 +50,67 @@ def test_table_dump(self):
         [self.assertEqual(expected_sqls[i], actual_sqls[i])
             for i in range(len(expected_sqls))]
 
+    def test_dump_autoincrement(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]
+        # the NULL value should now be automatically be set to 1
+        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 test_dump_autoincrement_create_new_db(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]
+        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 test_unorderable_row(self):
         # iterdump() should be able to cope with unorderable row types (issue #15545)
         class UnorderableRow:
@@ -70,6 +132,7 @@ def __getitem__(self, index):
         got = list(self.cx.iterdump())
         self.assertEqual(expected, got)
 
+
 def suite():
     tests = [
         DumpTests,
@@ -78,9 +141,11 @@ def suite():
         [unittest.TestLoader().loadTestsFromTestCase(t) for t in tests]
     )
 
+
 def test():
     runner = unittest.TextTestRunner()
     runner.run(suite())
 
+
 if __name__ == "__main__":
     test()

Misc/ACKS

@@ -922,6 +922,7 @@ Ron Klatchko
 Reid Kleckner
 Carsten Klein
 Bastian Kleineidam
+Joel Klimont
 Bob Kline
 Matthias Klose
 Jeremy Kloth

Misc/NEWS.d/next/Library/2018-09-28-22-18-03.bpo-34828.5Zyi_S.rst

@@ -0,0 +1,3 @@
+Sqlite3.iterdump now correctly deletes the "sqlite_sequence" table and
+inserts the rows from the dumped db. Just like the command ".dump" in
+sqlite3 does.