return '"{0}"'.format(name.replace('"', '""'))
+def _escape_single_quotes(value):
+ return value.replace("'", "''")
+
+
def _quote_value(value):
- return "'{0}'".format(value.replace("'", "''"))
+ return "'{0}'".format(_escape_single_quotes(value))
def _iterdump(connection, *, filter=None):
table_name_ident = _quote_name(table_name)
res = cu.execute(f'PRAGMA table_info({table_name_ident})')
column_names = [str(table_info[1]) for table_info in res.fetchall()]
- q = "SELECT 'INSERT INTO {0} VALUES('{1}')' FROM {0};".format(
- table_name_ident,
+ q = "SELECT 'INSERT INTO {0} VALUES('{1}')' FROM {2};".format(
+ _escape_single_quotes(table_name_ident),
"','".join(
"||quote({0})||".format(_quote_name(col)) for col in column_names
- )
+ ),
+ table_name_ident,
)
query_res = cu.execute(q)
for row in query_res:
[self.assertEqual(expected_sqls[i], actual_sqls[i])
for i in range(len(expected_sqls))]
+ def test_dump_single_quote_in_identifier(self):
+ # A single quote in a table or column name must not break the dump.
+ self.cu.execute("""CREATE TABLE "a'b" ("c'd" text);""")
+ self.cu.execute("""INSERT INTO "a'b" VALUES('x''y');""")
+ expected = [
+ "BEGIN TRANSACTION;",
+ """CREATE TABLE "a'b" ("c'd" text);""",
+ """INSERT INTO "a'b" VALUES('x''y');""",
+ "COMMIT;",
+ ]
+ actual = list(self.cx.iterdump())
+ self.assertEqual(expected, actual)
+ # The dump restores into a fresh database.
+ with memory_database() as cx2:
+ cx2.executescript("".join(actual))
+ row = cx2.execute("""SELECT "c'd" FROM "a'b";""").fetchone()
+ self.assertEqual(row[0], "x'y")
+
def test_table_dump_filter(self):
all_table_sqls = [
"""CREATE TABLE "some_table_2" ("id_1" INTEGER);""",