# future enhancements, you should normally quote any identifier that
# is an English language word, even if you do not have to."
-
-from contextlib import contextmanager
-
-
-def _force_decode(bs, *args, **kwargs):
- # gh-108590: Don't fail if the database contains invalid Unicode data.
- try:
- return bs.decode(*args, **kwargs)
- except UnicodeDecodeError:
- return "".join([chr(c) for c in bs])
-
-
-@contextmanager
-def _text_factory(con, factory):
- saved_factory = con.text_factory
- con.text_factory = factory
- try:
- yield
- finally:
- con.text_factory = saved_factory
-
-
def _iterdump(connection):
"""
Returns an iterator to the dump of the database in an SQL text format.
table_name_ident,
",".join("""'||quote("{0}")||'""".format(col.replace('"', '""')) for col in column_names))
query_res = cu.execute(q)
- with _text_factory(connection, bytes):
- for row in query_res:
- yield("{0};".format(_force_decode(row[0])))
+ for row in query_res:
+ yield("{0};".format(row[0]))
# Now when the type is 'index', 'trigger', or 'view'
q = """
actual = list(self.cx.iterdump())
self.assertEqual(expected, actual)
- def test_dump_unicode_invalid(self):
- # gh-108590
- expected = [
- "BEGIN TRANSACTION;",
- "CREATE TABLE foo (data TEXT);",
- "INSERT INTO \"foo\" VALUES('a\x9f');",
- "COMMIT;",
- ]
- self.cu.executescript("""
- CREATE TABLE foo (data TEXT);
- INSERT INTO foo VALUES (CAST(X'619f' AS TEXT));
- """)
- actual = list(self.cx.iterdump())
- self.assertEqual(expected, actual)
-
if __name__ == "__main__":
unittest.main()