Infinidoge/nomen
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

format and modify migrator to ignore internal sqlite tables

Browse source
This commit is contained in:
Infinidoge 2024-12-20 01:14:04 -05:00
parent 1d9c6e0155
commit f027f8c462
Signed by: Infinidoge
SSH key fingerprint: SHA256:GT2StvPQMMfFHyiiFJymQxfTG/z6EWLJ6NWItf5K5sA
1 changed files with 74 additions and 75 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

149
nomen/migrator.py
View file

@ -7,6 +7,8 @@ See <https://david.rothlis.net/declarative-schema-migration-for-sqlite>.
Author: William Manley <will@stb-tester.com>. Author: William Manley <will@stb-tester.com>.
Copyright © 2019-2022 Stb-tester.com Ltd. Copyright © 2019-2022 Stb-tester.com Ltd.
License: MIT. License: MIT.
Modified to ignore internal tables
""" """
import logging import logging
@ -14,6 +16,12 @@ import re
import sqlite3 import sqlite3
from textwrap import dedent from textwrap import dedent
INTERNAL_TABLES = {
"sqlite_stat1",
}
log = logging.getLogger("migrator")
def dumb_migrate_db(db, schema, allow_deletions=False): def dumb_migrate_db(db, schema, allow_deletions=False):
""" """
@ -77,21 +85,19 @@ class DBMigrator:
msg_argv += (args,) msg_argv += (args,)
else: else:
args = [] args = []
logging.info(msg_tmpl, *msg_argv) log.info(msg_tmpl, *msg_argv)
self.db.execute(sql, args) self.db.execute(sql, args)
self.n_changes += 1 self.n_changes += 1
def __enter__(self): def __enter__(self):
self.orig_foreign_keys = ( self.orig_foreign_keys = self.db.execute("PRAGMA foreign_keys").fetchone()[0]
self.db.execute("PRAGMA foreign_keys").fetchone()[0])
if self.orig_foreign_keys: if self.orig_foreign_keys:
self.log_execute("Disable foreign keys temporarily for migration", self.log_execute("Disable foreign keys temporarily for migration", "PRAGMA foreign_keys = OFF")
"PRAGMA foreign_keys = OFF")
# This doesn't count as a change because we'll undo it at the end # This doesn't count as a change because we'll undo it at the end
self.n_changes = 0 self.n_changes = 0
self.db.__enter__() self.db.__enter__()
self.db.execute('BEGIN') self.db.execute("BEGIN")
return self return self
def __exit__(self, exc_type, exc_value, exc_tb): def __exit__(self, exc_type, exc_value, exc_tb):
@ -103,7 +109,7 @@ class DBMigrator:
# > constraint enforcement may only be enabled or disabled when # > constraint enforcement may only be enabled or disabled when
# > there is no pending BEGIN or SAVEPOINT. # > there is no pending BEGIN or SAVEPOINT.
old_changes = self.n_changes old_changes = self.n_changes
new_val = self._migrate_pragma('foreign_keys') new_val = self._migrate_pragma("foreign_keys")
if new_val == self.orig_foreign_keys: if new_val == self.orig_foreign_keys:
self.n_changes = old_changes self.n_changes = old_changes
@ -115,142 +121,133 @@ class DBMigrator:
self.db.execute("VACUUM") self.db.execute("VACUUM")
else: else:
if self.orig_foreign_keys: if self.orig_foreign_keys:
self.log_execute( self.log_execute("Re-enable foreign keys after migration", "PRAGMA foreign_keys = ON")
"Re-enable foreign keys after migration",
"PRAGMA foreign_keys = ON")
def migrate(self): def migrate(self):
# In CI the database schema may be changing all the time. This checks # In CI the database schema may be changing all the time. This checks
# the current db and if it doesn't match database.sql we will # the current db and if it doesn't match database.sql we will
# modify it so it does match where possible. # modify it so it does match where possible.
pristine_tables = dict(self.pristine.execute("""\ pristine_tables = dict(
self.pristine.execute("""\
SELECT name, sql FROM sqlite_master SELECT name, sql FROM sqlite_master
WHERE type = \"table\" AND name != \"sqlite_sequence\"""").fetchall()) WHERE type = \"table\" AND name != \"sqlite_sequence\"""").fetchall()
pristine_indices = dict(self.pristine.execute("""\ )
pristine_indices = dict(
self.pristine.execute("""\
SELECT name, sql FROM sqlite_master SELECT name, sql FROM sqlite_master
WHERE type = \"index\"""").fetchall()) WHERE type = \"index\"""").fetchall()
)
tables = dict(self.db.execute("""\ tables = dict(
self.db.execute("""\
SELECT name, sql FROM sqlite_master SELECT name, sql FROM sqlite_master
WHERE type = \"table\" AND name != \"sqlite_sequence\"""").fetchall()) WHERE type = \"table\" AND name != \"sqlite_sequence\"""").fetchall()
)
new_tables = set(pristine_tables.keys()) - set(tables.keys()) new_tables = set(pristine_tables.keys()) - set(tables.keys()) - INTERNAL_TABLES
removed_tables = set(tables.keys()) - set(pristine_tables.keys()) removed_tables = set(tables.keys()) - set(pristine_tables.keys()) - INTERNAL_TABLES
if removed_tables and not self.allow_deletions: if removed_tables and not self.allow_deletions:
raise RuntimeError( raise RuntimeError("Database migration: Refusing to delete tables %r" % removed_tables)
"Database migration: Refusing to delete tables %r" %
removed_tables)
modified_tables = set( modified_tables = set(
name for name, sql in pristine_tables.items() name for name, sql in pristine_tables.items() if normalise_sql(tables.get(name, "")) != normalise_sql(sql)
if normalise_sql(tables.get(name, "")) != normalise_sql(sql)) )
# This PRAGMA is automatically disabled when the db is committed # This PRAGMA is automatically disabled when the db is committed
self.db.execute("PRAGMA defer_foreign_keys = TRUE") self.db.execute("PRAGMA defer_foreign_keys = TRUE")
# New and removed tables are easy: # New and removed tables are easy:
for tbl_name in new_tables: for tbl_name in new_tables:
self.log_execute("Create table %s" % tbl_name, self.log_execute("Create table %s" % tbl_name, pristine_tables[tbl_name])
pristine_tables[tbl_name])
for tbl_name in removed_tables: for tbl_name in removed_tables:
self.log_execute("Drop table %s" % tbl_name, self.log_execute("Drop table %s" % tbl_name, "DROP TABLE %s" % tbl_name)
"DROP TABLE %s" % tbl_name)
for tbl_name in modified_tables: for tbl_name in modified_tables:
# The SQLite documentation insists that we create the new table and # The SQLite documentation insists that we create the new table and
# rename it over the old rather than moving the old out of the way # rename it over the old rather than moving the old out of the way
# and then creating the new # and then creating the new
create_table_sql = pristine_tables[tbl_name] create_table_sql = pristine_tables[tbl_name]
create_table_sql = re.sub(r"\b%s\b" % re.escape(tbl_name), create_table_sql = re.sub(r"\b%s\b" % re.escape(tbl_name), tbl_name + "_migration_new", create_table_sql)
tbl_name + "_migration_new", self.log_execute("Columns change: Create table %s with updated schema" % tbl_name, create_table_sql)
create_table_sql)
self.log_execute(
"Columns change: Create table %s with updated schema" %
tbl_name, create_table_sql)
cols = set([ cols = set([x[1] for x in self.db.execute("PRAGMA table_info(%s)" % tbl_name)])
x[1] for x in self.db.execute( pristine_cols = set([x[1] for x in self.pristine.execute("PRAGMA table_info(%s)" % tbl_name)])
"PRAGMA table_info(%s)" % tbl_name)])
pristine_cols = set([
x[1] for x in
self.pristine.execute("PRAGMA table_info(%s)" % tbl_name)])
removed_columns = cols - pristine_cols removed_columns = cols - pristine_cols
if not self.allow_deletions and removed_columns: if not self.allow_deletions and removed_columns:
logging.warning( log.warning(
"Database migration: Refusing to remove columns %r from " "Database migration: Refusing to remove columns %r from "
"table %s. Current cols are %r attempting migration to %r", "table %s. Current cols are %r attempting migration to %r",
removed_columns, tbl_name, cols, pristine_cols) removed_columns,
tbl_name,
cols,
pristine_cols,
)
raise RuntimeError( raise RuntimeError(
"Database migration: Refusing to remove columns %r from " "Database migration: Refusing to remove columns %r from " "table %s" % (removed_columns, tbl_name)
"table %s" % (removed_columns, tbl_name)) )
logging.info("cols: %s, pristine_cols: %s", cols, pristine_cols) log.info("cols: %s, pristine_cols: %s", cols, pristine_cols)
self.log_execute( self.log_execute(
"Migrate data for table %s" % tbl_name, """\ "Migrate data for table %s" % tbl_name,
"""\
INSERT INTO {tbl_name}_migration_new ({common}) INSERT INTO {tbl_name}_migration_new ({common})
SELECT {common} FROM {tbl_name}""".format( SELECT {common} FROM {tbl_name}""".format(
tbl_name=tbl_name, tbl_name=tbl_name, common=", ".join(cols.intersection(pristine_cols))
common=", ".join(cols.intersection(pristine_cols)))) ),
)
# Don't need the old table any more # Don't need the old table any more
self.log_execute( self.log_execute("Drop old table %s now data has been migrated" % tbl_name, "DROP TABLE %s" % tbl_name)
"Drop old table %s now data has been migrated" % tbl_name,
"DROP TABLE %s" % tbl_name)
self.log_execute( self.log_execute(
"Columns change: Move new table %s over old" % tbl_name, "Columns change: Move new table %s over old" % tbl_name,
"ALTER TABLE %s_migration_new RENAME TO %s" % ( "ALTER TABLE %s_migration_new RENAME TO %s" % (tbl_name, tbl_name),
tbl_name, tbl_name)) )
# Migrate the indices # Migrate the indices
indices = dict(self.db.execute("""\ indices = dict(
self.db.execute("""\
SELECT name, sql FROM sqlite_master SELECT name, sql FROM sqlite_master
WHERE type = \"index\"""").fetchall()) WHERE type = \"index\"""").fetchall()
)
for name in set(indices.keys()) - set(pristine_indices.keys()): for name in set(indices.keys()) - set(pristine_indices.keys()):
self.log_execute("Dropping obsolete index %s" % name, self.log_execute("Dropping obsolete index %s" % name, "DROP INDEX %s" % name)
"DROP INDEX %s" % name)
for name, sql in pristine_indices.items(): for name, sql in pristine_indices.items():
if name not in indices: if name not in indices:
self.log_execute("Creating new index %s" % name, sql) self.log_execute("Creating new index %s" % name, sql)
elif sql != indices[name]: elif sql != indices[name]:
self.log_execute( self.log_execute("Index %s changed: Dropping old version" % name, "DROP INDEX %s" % name)
"Index %s changed: Dropping old version" % name, self.log_execute("Index %s changed: Creating updated version in its place" % name, sql)
"DROP INDEX %s" % name)
self.log_execute(
"Index %s changed: Creating updated version in its place" %
name, sql)
self._migrate_pragma('user_version') self._migrate_pragma("user_version")
if self.pristine.execute("PRAGMA foreign_keys").fetchone()[0]: if self.pristine.execute("PRAGMA foreign_keys").fetchone()[0]:
if self.db.execute("PRAGMA foreign_key_check").fetchall(): if self.db.execute("PRAGMA foreign_key_check").fetchall():
raise RuntimeError( raise RuntimeError("Database migration: Would fail foreign_key_check")
"Database migration: Would fail foreign_key_check")
def _migrate_pragma(self, pragma): def _migrate_pragma(self, pragma):
pristine_val = self.pristine.execute( pristine_val = self.pristine.execute("PRAGMA %s" % pragma).fetchone()[0]
"PRAGMA %s" % pragma).fetchone()[0]
val = self.db.execute("PRAGMA %s" % pragma).fetchone()[0] val = self.db.execute("PRAGMA %s" % pragma).fetchone()[0]
if val != pristine_val: if val != pristine_val:
self.log_execute( self.log_execute(
"Set %s to %i from %i" % (pragma, pristine_val, val), "Set %s to %i from %i" % (pragma, pristine_val, val), "PRAGMA %s = %i" % (pragma, pristine_val)
"PRAGMA %s = %i" % (pragma, pristine_val)) )
return pristine_val return pristine_val
def _left_pad(text, indent=" "): def _left_pad(text, indent=" "):
"""Maybe I can find a package in pypi for this?""" """Maybe I can find a package in pypi for this?"""
return "\n".join(indent + line for line in text.split('\n')) return "\n".join(indent + line for line in text.split("\n"))
def normalise_sql(sql): def normalise_sql(sql):
# Remove comments: # Remove comments:
sql = re.sub(r'--[^\n]*\n', "", sql) sql = re.sub(r"--[^\n]*\n", "", sql)
# Normalise whitespace: # Normalise whitespace:
sql = re.sub(r'\s+', " ", sql) sql = re.sub(r"\s+", " ", sql)
sql = re.sub(r" *([(),]) *", r"\1", sql) sql = re.sub(r" *([(),]) *", r"\1", sql)
# Remove unnecessary quotes # Remove unnecessary quotes
sql = re.sub(r'"(\w+)"', r"\1", sql) sql = re.sub(r'"(\w+)"', r"\1", sql)
@ -258,8 +255,10 @@ def normalise_sql(sql):
def test_normalise_sql(): def test_normalise_sql():
assert normalise_sql("""\ assert (
normalise_sql("""\
CREATE TABLE "Node"( -- This is my table CREATE TABLE "Node"( -- This is my table
-- There are many like it but this one is mine -- There are many like it but this one is mine
A b, C D, "E F G", h)""") == \ A b, C D, "E F G", h)""")
'CREATE TABLE Node(A b,C D,"E F G",h)' == 'CREATE TABLE Node(A b,C D,"E F G",h)'
)