Get add_foreign_keys() to work without modifying sqlite_master

[simonw]

sqlite-utils/sqlite_utils/db.py

Lines 1165 to 1174 in 13ebcc5

	cursor = self.conn.cursor()
	schema_version = cursor.execute("PRAGMA schema_version").fetchone()[0]
	cursor.execute("PRAGMA writable_schema = 1")
	for table_name, new_sql in table_sql.items():
	cursor.execute(
	"UPDATE sqlite_master SET sql = ? WHERE name = ?",
	(new_sql, table_name),
	)
	cursor.execute("PRAGMA schema_version = %d" % (schema_version + 1))
	cursor.execute("PRAGMA writable_schema = 0")

This is the only place in the code that attempts to modify sqlite_master directly, which fails on some Python installations.

Could this use the .transform() trick instead?

Or automatically switch to that trick if it hits an error?

[simonw]

I'm certain this could work.

It turns out the .transform() method already has code that creates the new table with a copy of foreign keys from the old one - dropping any foreign keys that were specified in the drop_foreign_keys= parameter:

sqlite-utils/sqlite_utils/db.py

Lines 1850 to 1872 in 1dc6b5a

	# foreign_keys
	create_table_foreign_keys = []
	for table, column, other_table, other_column in self.foreign_keys:
	if (drop_foreign_keys is None) or (column not in drop_foreign_keys):
	create_table_foreign_keys.append(
	(rename.get(column) or column, other_table, other_column)
	)
	if column_order is not None:
	column_order = [rename.get(col) or col for col in column_order]
	sqls = []
	sqls.append(
	self.db.create_table_sql(
	new_table_name,
	dict(new_column_pairs),
	pk=pk,
	not_null=create_table_not_null,
	defaults=create_table_defaults,
	foreign_keys=create_table_foreign_keys,
	column_order=column_order,
	).strip()
	)

Improving this code to support adding foreign keys as well would be pretty simple.

And then the .add_foreign_keys() and .add_foreign_key() methods could be updated to use .transform(...) under the hood instead.

[simonw]

This would help address these issues, among potentially many others:

• sqlite3.OperationalError: table sqlite_master may not be modified llm#60
• 'sqlite3.OperationalError: unable to open database file' error with a fresh install llm#116
• Unable to open database file llm#123

[simonw]

Looking at the whole of the .add_foreign_keys() method, the first section of it can remain unchanged - it's just a bunch of validation:

sqlite-utils/sqlite_utils/db.py

Lines 1106 to 1149 in 13ebcc5

	def add_foreign_keys(self, foreign_keys: Iterable[Tuple[str, str, str, str]]):
	"""
	See :ref:`python_api_add_foreign_keys`.
	:param foreign_keys: A list of ``(table, column, other_table, other_column)``
	tuples
	"""
	# foreign_keys is a list of explicit 4-tuples
	assert all(
	len(fk) == 4 and isinstance(fk, (list, tuple)) for fk in foreign_keys
	), "foreign_keys must be a list of 4-tuples, (table, column, other_table, other_column)"
	foreign_keys_to_create = []
	# Verify that all tables and columns exist
	for table, column, other_table, other_column in foreign_keys:
	if not self[table].exists():
	raise AlterError("No such table: {}".format(table))
	table_obj = self[table]
	if not isinstance(table_obj, Table):
	raise AlterError("Must be a table, not a view: {}".format(table))
	table_obj = cast(Table, table_obj)
	if column not in table_obj.columns_dict:
	raise AlterError("No such column: {} in {}".format(column, table))
	if not self[other_table].exists():
	raise AlterError("No such other_table: {}".format(other_table))
	if (
	other_column != "rowid"
	and other_column not in self[other_table].columns_dict
	):
	raise AlterError(
	"No such other_column: {} in {}".format(other_column, other_table)
	)
	# We will silently skip foreign keys that exist already
	if not any(
	fk
	for fk in table_obj.foreign_keys
	if fk.column == column
	and fk.other_table == other_table
	and fk.other_column == other_column
	):
	foreign_keys_to_create.append(
	(table, column, other_table, other_column)
	)

At that point we have foreign_keys_to_create as the ones that are new, but we should instead try to build up a foreign_keys which is both new and old, ready to be passed to .transform().

Here's the rest of that function, which will be replaced by a called to .transform(foreign_keys=foreign_keys):

sqlite-utils/sqlite_utils/db.py

Lines 1151 to 1177 in 13ebcc5

	# Construct SQL for use with "UPDATE sqlite_master SET sql = ? WHERE name = ?"
	table_sql: Dict[str, str] = {}
	for table, column, other_table, other_column in foreign_keys_to_create:
	old_sql = table_sql.get(table, self[table].schema)
	extra_sql = ",\n FOREIGN KEY([{column}]) REFERENCES [{other_table}]([{other_column}])\n".format(
	column=column, other_table=other_table, other_column=other_column
	)
	# Stick that bit in at the very end just before the closing ')'
	last_paren = old_sql.rindex(")")
	new_sql = old_sql[:last_paren].strip() + extra_sql + old_sql[last_paren:]
	table_sql[table] = new_sql
	# And execute it all within a single transaction
	with self.conn:
	cursor = self.conn.cursor()
	schema_version = cursor.execute("PRAGMA schema_version").fetchone()[0]
	cursor.execute("PRAGMA writable_schema = 1")
	for table_name, new_sql in table_sql.items():
	cursor.execute(
	"UPDATE sqlite_master SET sql = ? WHERE name = ?",
	(new_sql, table_name),
	)
	cursor.execute("PRAGMA schema_version = %d" % (schema_version + 1))
	cursor.execute("PRAGMA writable_schema = 0")
	# Have to VACUUM outside the transaction to ensure .foreign_keys property
	# can see the newly created foreign key.
	self.vacuum()

[simonw]

Actually I think table.transform() might get the following optional arguments:

    def transform(
        self,
        *,
        # ...
        # This one exists already:
        drop_foreign_keys: Optional[Iterable] = None,
        # These two are new. This one specifies keys to add:
        add_foreign_keys: Optional[ForeignKeysType] = None,
        # Or this one causes them all to be replaced with the new definitions:
        foreign_keys: Optional[ForeignKeysType] = None,

There should be validation that forbids you from using foreign_keys= at the same time as either drop_foreign_keys= or add_foreign_keys= because the point of foreign_keys= is to define the keys for the new table all in one go.

[simonw]

Maybe this:

        drop_foreign_keys: Optional[Iterable] = None,

Should be this:

        drop_foreign_keys: Optional[Iterable[str]] = None,

Because it takes a list of column names that should have their foreign keys dropped.

[simonw]

As a reminder:

sqlite-utils/sqlite_utils/db.py

Lines 159 to 165 in 1dc6b5a

	ForeignKeysType = Union[
	Iterable[str],
	Iterable[ForeignKey],
	Iterable[Tuple[str, str]],
	Iterable[Tuple[str, str, str]],
	Iterable[Tuple[str, str, str, str]],
	]

[simonw]

I'm inclined to just go with the .transform() method and not attempt to keep around the method that involves updating sqlite_master and then add code to detect if that's possible (or catch if it fails) and fall back on the other mechanism.

It would be nice to drop some code complexity, plus I don't yet have a way of running automated tests against Python + SQLite versions that exhibit the problem.

[simonw]

An interesting side-effect of this change is that it does result in a slightly different schema - e.g. this test:

sqlite-utils/tests/test_extract.py

Lines 118 to 133 in 1dc6b5a

	def test_extract_rowid_table(fresh_db):
	fresh_db["tree"].insert(
	{
	"name": "Tree 1",
	"common_name": "Palm",
	"latin_name": "Arecaceae",
	}
	)
	fresh_db["tree"].extract(["common_name", "latin_name"])
	assert fresh_db["tree"].schema == (
	'CREATE TABLE "tree" (\n'
	" [name] TEXT,\n"
	" [common_name_latin_name_id] INTEGER,\n"
	" FOREIGN KEY([common_name_latin_name_id]) REFERENCES [common_name_latin_name]([id])\n"
	")"
	)

Needs updating like so:

diff --git a/tests/test_extract.py b/tests/test_extract.py
index 70ad0cf..fd52534 100644
--- a/tests/test_extract.py
+++ b/tests/test_extract.py
@@ -127,8 +127,7 @@ def test_extract_rowid_table(fresh_db):
     assert fresh_db["tree"].schema == (
         'CREATE TABLE "tree" (\n'
         "   [name] TEXT,\n"
-        "   [common_name_latin_name_id] INTEGER,\n"
-        "   FOREIGN KEY([common_name_latin_name_id]) REFERENCES [common_name_latin_name]([id])\n"
+        "   [common_name_latin_name_id] INTEGER REFERENCES [common_name_latin_name]([id])\n"
         ")"
     )
     assert (

Unfortunately this means it may break other test suites that depend on sqlite-utils that have schema tests like this baked in.

I don't think this should count as a breaking change release though, but it's still worth noting.

[simonw]

Here's a new plugin that brings back the sqlite_master modifying version, for those that can use it:

https://github.com/simonw/sqlite-utils-fast-fks