directly but instead called from the Connection method, iterdump().
"""
+ writeable_schema = False
cu = connection.cursor()
yield('BEGIN TRANSACTION;')
yield('ANALYZE "sqlite_master";')
elif table_name.startswith('sqlite_'):
continue
- # NOTE: Virtual table support not implemented
- #elif sql.startswith('CREATE VIRTUAL TABLE'):
- # qtable = table_name.replace("'", "''")
- # yield("INSERT INTO sqlite_master(type,name,tbl_name,rootpage,sql)"\
- # "VALUES('table','{0}','{0}',0,'{1}');".format(
- # qtable,
- # sql.replace("''")))
+ elif sql.startswith('CREATE VIRTUAL TABLE'):
+ if not writeable_schema:
+ writeable_schema = True
+ yield('PRAGMA writable_schema=ON;')
+ yield("INSERT INTO sqlite_master(type,name,tbl_name,rootpage,sql)"
+ "VALUES('table',{0},{0},0,{1});".format(
+ _quote_value(table_name),
+ _quote_value(sql),
+ ))
else:
yield('{0};'.format(sql))
for name, type, sql in schema_res.fetchall():
yield('{0};'.format(sql))
+ if writeable_schema:
+ yield('PRAGMA writable_schema=OFF;')
+
# gh-79009: Yield statements concerning the sqlite_sequence table at the
# end of the transaction.
for row in sqlite_sequence:
got = list(self.cx.iterdump())
self.assertEqual(expected, got)
+ def test_dump_virtual_tables(self):
+ # gh-64662
+ expected = [
+ "BEGIN TRANSACTION;",
+ "PRAGMA writable_schema=ON;",
+ ("INSERT INTO sqlite_master(type,name,tbl_name,rootpage,sql)"
+ "VALUES('table','test','test',0,'CREATE VIRTUAL TABLE test USING fts4(example)');"),
+ "CREATE TABLE 'test_content'(docid INTEGER PRIMARY KEY, 'c0example');",
+ "CREATE TABLE 'test_docsize'(docid INTEGER PRIMARY KEY, size BLOB);",
+ ("CREATE TABLE 'test_segdir'(level INTEGER,idx INTEGER,start_block INTEGER,"
+ "leaves_end_block INTEGER,end_block INTEGER,root BLOB,PRIMARY KEY(level, idx));"),
+ "CREATE TABLE 'test_segments'(blockid INTEGER PRIMARY KEY, block BLOB);",
+ "CREATE TABLE 'test_stat'(id INTEGER PRIMARY KEY, value BLOB);",
+ "PRAGMA writable_schema=OFF;",
+ "COMMIT;"
+ ]
+ self.cu.execute("CREATE VIRTUAL TABLE test USING fts4(example)")
+ actual = list(self.cx.iterdump())
+ self.assertEqual(expected, actual)
+
if __name__ == "__main__":
unittest.main()