mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-03 08:34:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

gh-91602: Add iterdump() support for filtering database objects (#114501)

Browse source 
Add optional 'filter' parameter to iterdump() that allows a "LIKE"
pattern for filtering database objects to dump.

Co-authored-by: Erlend E. Aasland <erlend@python.org>
This commit is contained in:
Mariusz Felisiak 2024-02-06 12:34:56 +01:00 • committed by GitHub
parent 4bf41879d0
commit 1a10437a14
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 176 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

70
Lib/test/test_sqlite3/test_dump.py
View file

@ -54,6 +54,76 @@ class DumpTests(MemoryDatabaseMixin, unittest.TestCase):
[self.assertEqual(expected_sqls[i], actual_sqls[i])
for i in range(len(expected_sqls))]
def test_table_dump_filter(self):
all_table_sqls = [
"""CREATE TABLE "some_table_2" ("id_1" INTEGER);""",
"""INSERT INTO "some_table_2" VALUES(3);""",
"""INSERT INTO "some_table_2" VALUES(4);""",
"""CREATE TABLE "test_table_1" ("id_2" INTEGER);""",
"""INSERT INTO "test_table_1" VALUES(1);""",
"""INSERT INTO "test_table_1" VALUES(2);""",
]
all_views_sqls = [
"""CREATE VIEW "view_1" AS SELECT * FROM "some_table_2";""",
"""CREATE VIEW "view_2" AS SELECT * FROM "test_table_1";""",
]
# Create database structure.
for sql in [*all_table_sqls, *all_views_sqls]:
self.cu.execute(sql)
# %_table_% matches all tables.
dump_sqls = list(self.cx.iterdump(filter="%_table_%"))
self.assertEqual(
dump_sqls,
["BEGIN TRANSACTION;", *all_table_sqls, "COMMIT;"],
)
# view_% matches all views.
dump_sqls = list(self.cx.iterdump(filter="view_%"))
self.assertEqual(
dump_sqls,
["BEGIN TRANSACTION;", *all_views_sqls, "COMMIT;"],
)
# %_1 matches tables and views with the _1 suffix.
dump_sqls = list(self.cx.iterdump(filter="%_1"))
self.assertEqual(
dump_sqls,
[
"BEGIN TRANSACTION;",
"""CREATE TABLE "test_table_1" ("id_2" INTEGER);""",
"""INSERT INTO "test_table_1" VALUES(1);""",
"""INSERT INTO "test_table_1" VALUES(2);""",
"""CREATE VIEW "view_1" AS SELECT * FROM "some_table_2";""",
"COMMIT;"
],
)
# some_% matches some_table_2.
dump_sqls = list(self.cx.iterdump(filter="some_%"))
self.assertEqual(
dump_sqls,
[
"BEGIN TRANSACTION;",
"""CREATE TABLE "some_table_2" ("id_1" INTEGER);""",
"""INSERT INTO "some_table_2" VALUES(3);""",
"""INSERT INTO "some_table_2" VALUES(4);""",
"COMMIT;"
],
)
# Only single object.
dump_sqls = list(self.cx.iterdump(filter="view_2"))
self.assertEqual(
dump_sqls,
[
"BEGIN TRANSACTION;",
"""CREATE VIEW "view_2" AS SELECT * FROM "test_table_1";""",
"COMMIT;"
],
)
# % matches all objects.
dump_sqls = list(self.cx.iterdump(filter="%"))
self.assertEqual(
dump_sqls,
["BEGIN TRANSACTION;", *all_table_sqls, *all_views_sqls, "COMMIT;"],
)
def test_dump_autoincrement(self):
expected = [
'CREATE TABLE "t1" (id integer primary key autoincrement);',