Use .transform() instead of modifying sqlite_master for add_foreign_keys(), refs #577

sqlite_utils/db.py

@@ -747,9 +747,16 @@ class Database:
     def resolve_foreign_keys(
         self, name: str, foreign_keys: ForeignKeysType
     ) -> List[ForeignKey]:
-        # foreign_keys may be a list of column names, a list of ForeignKey tuples,
+        """
-        # a list of tuple-pairs or a list of tuple-triples. We want to turn
+        Given a list of differing foreign_keys definitions, return a list of
-        # it into a list of ForeignKey tuples
+        fully resolved ForeignKey() named tuples.
+
+        :param name: Name of table that foreign keys are being defined for
+        :param foreign_keys: List of foreign keys, each of which can be a
+            string, a ForeignKey() named tuple, a tuple of (column, other_table),
+            or a tuple of (column, other_table, other_column), or a tuple of
+            (table, column, other_table, other_column)
+        """
         table = cast(Table, self[name])
         if all(isinstance(fk, ForeignKey) for fk in foreign_keys):
             return cast(List[ForeignKey], foreign_keys)
@@ -767,6 +774,11 @@ class Database:
         ), "foreign_keys= should be a list of tuples"
         fks = []
         for tuple_or_list in foreign_keys:
+            if len(tuple_or_list) == 4:
+                assert (
+                    tuple_or_list[0] == name
+                ), "First item in {} should have been {}".format(tuple_or_list, name)
+                tuple_or_list = tuple(tuple_or_list[1:])
             assert len(tuple_or_list) in (
                 2,
                 3,
@@ -1148,32 +1160,14 @@ class Database:
                     (table, column, other_table, other_column)
                 )
 
-        # Construct SQL for use with "UPDATE sqlite_master SET sql = ? WHERE name = ?"
+        # Group them by table
-        table_sql: Dict[str, str] = {}
+        by_table = {}
-        for table, column, other_table, other_column in foreign_keys_to_create:
+        for fk in foreign_keys_to_create:
-            old_sql = table_sql.get(table, self[table].schema)
+            by_table.setdefault(fk[0], []).append(fk)
-            extra_sql = ",\n   FOREIGN KEY([{column}]) REFERENCES [{other_table}]([{other_column}])\n".format(
+
-                column=column, other_table=other_table, other_column=other_column
+        for table, fks in by_table.items():
-            )
+            self[table].transform(add_foreign_keys=fks)
-            # 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()
 
     def index_foreign_keys(self):
@@ -1704,7 +1698,9 @@ class Table(Queryable):
         pk: Optional[Any] = DEFAULT,
         not_null: Optional[Iterable[str]] = None,
         defaults: Optional[Dict[str, Any]] = None,
-        drop_foreign_keys: Optional[Iterable] = None,
+        drop_foreign_keys: Optional[Iterable[str]] = None,
+        add_foreign_keys: Optional[ForeignKeysType] = None,
+        foreign_keys: Optional[ForeignKeysType] = None,
         column_order: Optional[List[str]] = None,
         keep_table: Optional[str] = None,
     ) -> "Table":
@@ -1721,6 +1717,8 @@ class Table(Queryable):
         :param not_null: Columns to set as ``NOT NULL``
         :param defaults: Default values for columns
         :param drop_foreign_keys: Names of columns that should have their foreign key constraints removed
+        :param add_foreign_keys: List of foreign keys to add to the table
+        :param foreign_keys: List of foreign keys to set for the table, replacing any existing foreign keys
         :param column_order: List of strings specifying a full or partial column order
           to use when creating the table
         :param keep_table: If specified, the existing table will be renamed to this and will not be
@@ -1735,6 +1733,8 @@ class Table(Queryable):
             not_null=not_null,
             defaults=defaults,
             drop_foreign_keys=drop_foreign_keys,
+            add_foreign_keys=add_foreign_keys,
+            foreign_keys=foreign_keys,
             column_order=column_order,
             keep_table=keep_table,
         )
@@ -1765,6 +1765,8 @@ class Table(Queryable):
         not_null: Optional[Iterable[str]] = None,
         defaults: Optional[Dict[str, Any]] = None,
         drop_foreign_keys: Optional[Iterable] = None,
+        add_foreign_keys: Optional[ForeignKeysType] = None,
+        foreign_keys: Optional[ForeignKeysType] = None,
         column_order: Optional[List[str]] = None,
         tmp_suffix: Optional[str] = None,
         keep_table: Optional[str] = None,
@@ -1779,6 +1781,8 @@ class Table(Queryable):
         :param not_null: Columns to set as ``NOT NULL``
         :param defaults: Default values for columns
         :param drop_foreign_keys: Names of columns that should have their foreign key constraints removed
+        :param add_foreign_keys: List of foreign keys to add to the table
+        :param foreign_keys: List of foreign keys to set for the table, replacing any existing foreign keys
         :param column_order: List of strings specifying a full or partial column order
           to use when creating the table
         :param tmp_suffix: Suffix to use for the temporary table name
@@ -1788,6 +1792,43 @@ class Table(Queryable):
         types = types or {}
         rename = rename or {}
         drop = drop or set()
+
+        if foreign_keys is not None:
+            if add_foreign_keys is not None:
+                raise ValueError(
+                    "Cannot specify both foreign_keys and add_foreign_keys"
+                )
+            if drop_foreign_keys is not None:
+                raise ValueError(
+                    "Cannot specify both foreign_keys and drop_foreign_keys"
+                )
+            create_table_foreign_keys = foreign_keys
+        else:
+            # Construct foreign_keys from current, plus add_foreign_keys, minus drop_foreign_keys
+            create_table_foreign_keys = []
+            for table, column, other_table, other_column in self.foreign_keys:
+                # Copy over old foreign keys, unless we are dropping them
+                if (drop_foreign_keys is None) or (column not in drop_foreign_keys):
+                    create_table_foreign_keys.append(
+                        ForeignKey(
+                            table,
+                            rename.get(column) or column,
+                            other_table,
+                            other_column,
+                        )
+                    )
+            # Add new foreign keys
+            if add_foreign_keys is not None:
+                for fk in self.db.resolve_foreign_keys(self.name, add_foreign_keys):
+                    create_table_foreign_keys.append(
+                        ForeignKey(
+                            self.name,
+                            rename.get(fk.column) or fk.column,
+                            fk.other_table,
+                            fk.other_column,
+                        )
+                    )
+
         new_table_name = "{}_new_{}".format(
             self.name, tmp_suffix or os.urandom(6).hex()
         )
@@ -1847,14 +1888,6 @@ class Table(Queryable):
                 {rename.get(c) or c: v for c, v in defaults.items()}
             )
 
-        # 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]
 

tests/test_cli.py

@@ -192,7 +192,7 @@ def test_output_table(db_path, options, expected):
             ]
         )
     result = CliRunner().invoke(cli.cli, ["rows", db_path, "rows"] + options)
-    assert 0 == result.exit_code
+    assert result.exit_code == 0
     assert expected == result.output.strip()
 
 
@@ -200,7 +200,7 @@ def test_create_index(db_path):
     db = Database(db_path)
     assert [] == db["Gosh"].indexes
     result = CliRunner().invoke(cli.cli, ["create-index", db_path, "Gosh", "c1"])
-    assert 0 == result.exit_code
+    assert result.exit_code == 0
     assert [
         Index(
             seq=0, name="idx_Gosh_c1", unique=0, origin="c", partial=0, columns=["c1"]
@@ -210,7 +210,7 @@ def test_create_index(db_path):
     result = CliRunner().invoke(
         cli.cli, ["create-index", db_path, "Gosh", "c2", "--name", "blah"]
     )
-    assert 0 == result.exit_code
+    assert result.exit_code == 0
     assert [
         Index(seq=0, name="blah", unique=0, origin="c", partial=0, columns=["c2"]),
         Index(
@@ -227,7 +227,7 @@ def test_create_index(db_path):
         "--unique",
     ]
     result = CliRunner().invoke(cli.cli, create_index_unique_args)
-    assert 0 == result.exit_code
+    assert result.exit_code == 0
     assert [
         Index(
             seq=0,
@@ -366,7 +366,7 @@ def test_add_foreign_key(db_path, args, assert_message):
     result = CliRunner().invoke(
         cli.cli, ["add-foreign-key", db_path, "books", "author_id", "authors", "id"]
     )
-    assert 0 != result.exit_code
+    assert result.exit_code != 0
     assert (
         "Error: Foreign key already exists for author_id => authors.id"
         == result.output.strip()
@@ -377,13 +377,13 @@ def test_add_foreign_key(db_path, args, assert_message):
         cli.cli,
         ["add-foreign-key", db_path, "books", "author_id", "authors", "id", "--ignore"],
     )
-    assert 0 == result.exit_code
+    assert result.exit_code == 0
 
     # Error if we try against an invalid column
     result = CliRunner().invoke(
         cli.cli, ["add-foreign-key", db_path, "books", "author_id", "authors", "bad"]
     )
-    assert 0 != result.exit_code
+    assert result.exit_code != 0
     assert "Error: No such column: authors.bad" == result.output.strip()
 
 
@@ -395,10 +395,12 @@ def test_add_column_foreign_key(db_path):
     result = CliRunner().invoke(
         cli.cli, ["add-column", db_path, "books", "author_id", "--fk", "authors"]
     )
-    assert 0 == result.exit_code, result.output
+    assert result.exit_code == 0, result.output
-    assert (
+    assert db["books"].schema == (
-        "CREATE TABLE [books] ( [title] TEXT , [author_id] INTEGER, FOREIGN KEY([author_id]) REFERENCES [authors]([id]) )"
+        'CREATE TABLE "books" (\n'
-        == collapse_whitespace(db["books"].schema)
+        "   [title] TEXT,\n"
+        "   [author_id] INTEGER REFERENCES [authors]([id])\n"
+        ")"
     )
     # Try it again with a custom --fk-col
     result = CliRunner().invoke(
@@ -414,18 +416,19 @@ def test_add_column_foreign_key(db_path):
             "name",
         ],
     )
-    assert 0 == result.exit_code, result.output
+    assert result.exit_code == 0, result.output
-    assert (
+    assert db["books"].schema == (
-        "CREATE TABLE [books] ( [title] TEXT , [author_id] INTEGER, [author_name_ref] TEXT, "
+        'CREATE TABLE "books" (\n'
-        "FOREIGN KEY([author_id]) REFERENCES [authors]([id]), "
+        "   [title] TEXT,\n"
-        "FOREIGN KEY([author_name_ref]) REFERENCES [authors]([name]) )"
+        "   [author_id] INTEGER REFERENCES [authors]([id]),\n"
-        == collapse_whitespace(db["books"].schema)
+        "   [author_name_ref] TEXT REFERENCES [authors]([name])\n"
+        ")"
     )
     # Throw an error if the --fk table does not exist
     result = CliRunner().invoke(
         cli.cli, ["add-column", db_path, "books", "author_id", "--fk", "bobcats"]
     )
-    assert 0 != result.exit_code
+    assert result.exit_code != 0
     assert "table 'bobcats' does not exist" in str(result.exception)
 
 
@@ -449,7 +452,7 @@ def test_index_foreign_keys(db_path):
     db = Database(db_path)
     assert [] == db["books"].indexes
     result = CliRunner().invoke(cli.cli, ["index-foreign-keys", db_path])
-    assert 0 == result.exit_code
+    assert result.exit_code == 0
     assert [["author_id"], ["author_name_ref"]] == [
         i.columns for i in db["books"].indexes
     ]
@@ -461,7 +464,7 @@ def test_enable_fts(db_path):
     result = CliRunner().invoke(
         cli.cli, ["enable-fts", db_path, "Gosh", "c1", "--fts4"]
     )
-    assert 0 == result.exit_code
+    assert result.exit_code == 0
     assert "Gosh_fts" == db["Gosh"].detect_fts()
 
     # Table names with restricted chars are handled correctly.
@@ -480,7 +483,7 @@ def test_enable_fts(db_path):
             "porter",
         ],
     )
-    assert 0 == result.exit_code
+    assert result.exit_code == 0
     assert "http://example.com_fts" == db["http://example.com"].detect_fts()
     # Check tokenize was set to porter
     assert (
@@ -528,7 +531,7 @@ def test_enable_fts_with_triggers(db_path):
         )
         .exit_code
     )
-    assert 0 == exit_code
+    assert exit_code == 0
 
     def search(q):
         return (
@@ -549,7 +552,7 @@ def test_populate_fts(db_path):
         .invoke(cli.cli, ["enable-fts", db_path, "Gosh", "c1", "--fts4"])
         .exit_code
     )
-    assert 0 == exit_code
+    assert exit_code == 0
 
     def search(q):
         return (
@@ -564,7 +567,7 @@ def test_populate_fts(db_path):
     exit_code = (
         CliRunner().invoke(cli.cli, ["populate-fts", db_path, "Gosh", "c1"]).exit_code
     )
-    assert 0 == exit_code
+    assert exit_code == 0
     assert [("martha",)] == search("martha")
 
 
@@ -582,13 +585,13 @@ def test_disable_fts(db_path):
         "Gosh_fts_docsize",
     } == set(db.table_names())
     exit_code = CliRunner().invoke(cli.cli, ["disable-fts", db_path, "Gosh"]).exit_code
-    assert 0 == exit_code
+    assert exit_code == 0
     assert {"Gosh", "Gosh2"} == set(db.table_names())
 
 
 def test_vacuum(db_path):
     result = CliRunner().invoke(cli.cli, ["vacuum", db_path])
-    assert 0 == result.exit_code
+    assert result.exit_code == 0
 
 
 def test_dump(db_path):
@@ -617,7 +620,7 @@ def test_optimize(db_path, tables):
         db["Gosh2"].enable_fts(["c1", "c2", "c3"], fts_version="FTS5")
     size_before_optimize = os.stat(db_path).st_size
     result = CliRunner().invoke(cli.cli, ["optimize", db_path] + tables)
-    assert 0 == result.exit_code
+    assert result.exit_code == 0
     size_after_optimize = os.stat(db_path).st_size
     # Weirdest thing: tests started failing because size after
     # ended up larger than size before in some cases. I think
@@ -625,7 +628,7 @@ def test_optimize(db_path, tables):
     assert size_after_optimize <= (size_before_optimize + 10000)
     # Soundness check that --no-vacuum doesn't throw errors:
     result = CliRunner().invoke(cli.cli, ["optimize", "--no-vacuum", db_path])
-    assert 0 == result.exit_code
+    assert result.exit_code == 0
 
 
 def test_rebuild_fts_fixes_docsize_error(db_path):
@@ -653,7 +656,7 @@ def test_rebuild_fts_fixes_docsize_error(db_path):
     assert db["fts5_table_fts_docsize"].count == 20000
     # Running rebuild-fts should fix this
     result = CliRunner().invoke(cli.cli, ["rebuild-fts", db_path, "fts5_table"])
-    assert 0 == result.exit_code
+    assert result.exit_code == 0
     assert db["fts5_table_fts_docsize"].count == 10000
 
 
@@ -676,7 +679,7 @@ def test_query_csv(db_path, format, expected):
     result = CliRunner().invoke(
         cli.cli, [db_path, "select id, name, age from dogs", format]
     )
-    assert 0 == result.exit_code
+    assert result.exit_code == 0
     assert result.output.replace("\r", "") == expected
     # Test the no-headers option:
     result = CliRunner().invoke(
@@ -1027,7 +1030,7 @@ def test_upsert(db_path, tmpdir):
         ["insert", db_path, "dogs", json_path, "--pk", "id"],
         catch_exceptions=False,
     )
-    assert 0 == result.exit_code, result.output
+    assert result.exit_code == 0, result.output
     assert 2 == db["dogs"].count
     # Now run the upsert to update just their ages
     upsert_dogs = [
@@ -1040,7 +1043,7 @@ def test_upsert(db_path, tmpdir):
         ["upsert", db_path, "dogs", json_path, "--pk", "id"],
         catch_exceptions=False,
     )
-    assert 0 == result.exit_code, result.output
+    assert result.exit_code == 0, result.output
     assert list(db.query("select * from dogs order by id")) == [
         {"id": 1, "name": "Cleo", "age": 5},
         {"id": 2, "name": "Nixie", "age": 5},
@@ -1073,7 +1076,7 @@ def test_upsert_analyze(db_path, tmpdir):
         ["upsert", db_path, "rows", "-", "--nl", "--analyze", "--pk", "id"],
         input='{"id": 2, "foo": "bar", "n": 1}',
     )
-    assert 0 == result.exit_code, result.output
+    assert result.exit_code == 0, result.output
     assert "sqlite_stat1" in db.table_names()
 
 
@@ -1100,7 +1103,7 @@ def test_upsert_alter(db_path, tmpdir):
     result = CliRunner().invoke(
         cli.cli, ["insert", db_path, "dogs", json_path, "--pk", "id"]
     )
-    assert 0 == result.exit_code, result.output
+    assert result.exit_code == 0, result.output
     # Should fail with error code if no --alter
     upsert_dogs = [{"id": 1, "age": 5}]
     write_json(json_path, upsert_dogs)
@@ -1117,7 +1120,7 @@ def test_upsert_alter(db_path, tmpdir):
     result = CliRunner().invoke(
         cli.cli, ["upsert", db_path, "dogs", json_path, "--pk", "id", "--alter"]
     )
-    assert 0 == result.exit_code
+    assert result.exit_code == 0
     assert [
         {"id": 1, "name": "Cleo", "age": 5},
     ] == list(db.query("select * from dogs order by id"))
@@ -1187,7 +1190,7 @@ def test_create_table(args, schema):
             + args,
             catch_exceptions=False,
         )
-        assert 0 == result.exit_code
+        assert result.exit_code == 0
         db = Database("test.db")
         assert schema == db["t"].schema
 
@@ -1217,7 +1220,7 @@ def test_create_table_foreign_key():
             result = runner.invoke(
                 cli.cli, ["create-table", "books.db"] + args, catch_exceptions=False
             )
-            assert 0 == result.exit_code
+            assert result.exit_code == 0
         db = Database("books.db")
         assert (
             "CREATE TABLE [authors] (\n"
@@ -1257,7 +1260,7 @@ def test_create_table_ignore():
         result = runner.invoke(
             cli.cli, ["create-table", "test.db", "dogs", "id", "integer", "--ignore"]
         )
-        assert 0 == result.exit_code
+        assert result.exit_code == 0
         assert "CREATE TABLE [dogs] (\n   [name] TEXT\n)" == db["dogs"].schema
 
 
@@ -1269,7 +1272,7 @@ def test_create_table_replace():
         result = runner.invoke(
             cli.cli, ["create-table", "test.db", "dogs", "id", "integer", "--replace"]
         )
-        assert 0 == result.exit_code
+        assert result.exit_code == 0
         assert "CREATE TABLE [dogs] (\n   [id] INTEGER\n)" == db["dogs"].schema
 
 
@@ -1280,7 +1283,7 @@ def test_create_view():
         result = runner.invoke(
             cli.cli, ["create-view", "test.db", "version", "select sqlite_version()"]
         )
-        assert 0 == result.exit_code
+        assert result.exit_code == 0
         assert "CREATE VIEW version AS select sqlite_version()" == db["version"].schema
 
 
@@ -1314,7 +1317,7 @@ def test_create_view_ignore():
                 "--ignore",
             ],
         )
-        assert 0 == result.exit_code
+        assert result.exit_code == 0
         assert (
             "CREATE VIEW version AS select sqlite_version() + 1" == db["version"].schema
         )
@@ -1335,7 +1338,7 @@ def test_create_view_replace():
                 "--replace",
             ],
         )
-        assert 0 == result.exit_code
+        assert result.exit_code == 0
         assert "CREATE VIEW version AS select sqlite_version()" == db["version"].schema
 
 
@@ -1353,7 +1356,7 @@ def test_drop_table():
                 "t",
             ],
         )
-        assert 0 == result.exit_code
+        assert result.exit_code == 0
         assert "t" not in db.table_names()
 
 
@@ -1377,7 +1380,7 @@ def test_drop_table_error():
             cli.cli,
             ["drop-table", "test.db", "t2", "--ignore"],
         )
-        assert 0 == result.exit_code
+        assert result.exit_code == 0
 
 
 def test_drop_view():
@@ -1394,7 +1397,7 @@ def test_drop_view():
                 "hello",
             ],
         )
-        assert 0 == result.exit_code
+        assert result.exit_code == 0
         assert "hello" not in db.view_names()
 
 
@@ -1418,7 +1421,7 @@ def test_drop_view_error():
             cli.cli,
             ["drop-view", "test.db", "t2", "--ignore"],
         )
-        assert 0 == result.exit_code
+        assert result.exit_code == 0
 
 
 def test_enable_wal():
@@ -1430,7 +1433,7 @@ def test_enable_wal():
             db["t"].create({"pk": int}, pk="pk")
             assert db.journal_mode == "delete"
         result = runner.invoke(cli.cli, ["enable-wal"] + dbs, catch_exceptions=False)
-        assert 0 == result.exit_code
+        assert result.exit_code == 0
         for dbname in dbs:
             db = Database(dbname)
             assert db.journal_mode == "wal"
@@ -1446,7 +1449,7 @@ def test_disable_wal():
             db.enable_wal()
             assert db.journal_mode == "wal"
         result = runner.invoke(cli.cli, ["disable-wal"] + dbs)
-        assert 0 == result.exit_code
+        assert result.exit_code == 0
         for dbname in dbs:
             db = Database(dbname)
             assert db.journal_mode == "delete"
@@ -1700,8 +1703,7 @@ _common_other_schema = (
                 'CREATE TABLE "trees" (\n'
                 "   [id] INTEGER PRIMARY KEY,\n"
                 "   [address] TEXT,\n"
-                "   [species_id] INTEGER,\n"
+                "   [species_id] INTEGER REFERENCES [species]([id])\n"
-                "   FOREIGN KEY([species_id]) REFERENCES [species]([id])\n"
                 ")"
             ),
             _common_other_schema,
@@ -1712,8 +1714,7 @@ _common_other_schema = (
                 'CREATE TABLE "trees" (\n'
                 "   [id] INTEGER PRIMARY KEY,\n"
                 "   [address] TEXT,\n"
-                "   [custom_table_id] INTEGER,\n"
+                "   [custom_table_id] INTEGER REFERENCES [custom_table]([id])\n"
-                "   FOREIGN KEY([custom_table_id]) REFERENCES [custom_table]([id])\n"
                 ")"
             ),
             "CREATE TABLE [custom_table] (\n   [id] INTEGER PRIMARY KEY,\n   [species] TEXT\n)",
@@ -1724,8 +1725,7 @@ _common_other_schema = (
                 'CREATE TABLE "trees" (\n'
                 "   [id] INTEGER PRIMARY KEY,\n"
                 "   [address] TEXT,\n"
-                "   [custom_fk] INTEGER,\n"
+                "   [custom_fk] INTEGER REFERENCES [species]([id])\n"
-                "   FOREIGN KEY([custom_fk]) REFERENCES [species]([id])\n"
                 ")"
             ),
             _common_other_schema,
@@ -1735,8 +1735,7 @@ _common_other_schema = (
             'CREATE TABLE "trees" (\n'
             "   [id] INTEGER PRIMARY KEY,\n"
             "   [address] TEXT,\n"
-            "   [species_id] INTEGER,\n"
+            "   [species_id] INTEGER REFERENCES [species]([id])\n"
-            "   FOREIGN KEY([species_id]) REFERENCES [species]([id])\n"
             ")",
             "CREATE TABLE [species] (\n   [id] INTEGER PRIMARY KEY,\n   [species] TEXT\n)",
         ),

tests/test_create.py

@@ -490,8 +490,8 @@ def test_add_foreign_key_guess_table(fresh_db):
     fresh_db["dogs"].add_column("breed_id", int)
     fresh_db["dogs"].add_foreign_key("breed_id")
     assert (
-        "CREATE TABLE [dogs] ( [name] TEXT , [breed_id] INTEGER, FOREIGN KEY([breed_id]) REFERENCES [breeds]([id]) )"
+        collapse_whitespace(fresh_db["dogs"].schema)
-        == collapse_whitespace(fresh_db["dogs"].schema)
+        == 'CREATE TABLE "dogs" ( [name] TEXT, [breed_id] INTEGER REFERENCES [breeds]([id]) )'
     )
 
 

tests/test_extract.py

@@ -26,11 +26,8 @@ def test_extract_single_column(fresh_db, table, fk_column):
         'CREATE TABLE "tree" (\n'
         "   [id] INTEGER PRIMARY KEY,\n"
         "   [name] TEXT,\n"
-        "   [{}] INTEGER,\n".format(expected_fk)
+        "   [{}] INTEGER REFERENCES [{}]([id]),\n".format(expected_fk, expected_table)
-        + "   [end] INTEGER,\n"
+        + "   [end] INTEGER\n"
-        + "   FOREIGN KEY([{}]) REFERENCES [{}]([id])\n".format(
-            expected_fk, expected_table
-        )
         + ")"
     )
     assert fresh_db[expected_table].schema == (
@@ -76,8 +73,7 @@ def test_extract_multiple_columns_with_rename(fresh_db):
         'CREATE TABLE "tree" (\n'
         "   [id] INTEGER PRIMARY KEY,\n"
         "   [name] TEXT,\n"
-        "   [common_name_latin_name_id] INTEGER,\n"
+        "   [common_name_latin_name_id] INTEGER REFERENCES [common_name_latin_name]([id])\n"
-        "   FOREIGN KEY([common_name_latin_name_id]) REFERENCES [common_name_latin_name]([id])\n"
         ")"
     )
     assert fresh_db["common_name_latin_name"].schema == (
@@ -127,8 +123,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"
+        "   [common_name_latin_name_id] INTEGER REFERENCES [common_name_latin_name]([id])\n"
-        "   FOREIGN KEY([common_name_latin_name_id]) REFERENCES [common_name_latin_name]([id])\n"
         ")"
     )
     assert (
@@ -158,16 +153,14 @@ def test_reuse_lookup_table(fresh_db):
     assert fresh_db["sightings"].schema == (
         'CREATE TABLE "sightings" (\n'
         "   [id] INTEGER PRIMARY KEY,\n"
-        "   [species_id] INTEGER,\n"
+        "   [species_id] INTEGER REFERENCES [species]([id])\n"
-        "   FOREIGN KEY([species_id]) REFERENCES [species]([id])\n"
         ")"
     )
     assert fresh_db["individuals"].schema == (
         'CREATE TABLE "individuals" (\n'
         "   [id] INTEGER PRIMARY KEY,\n"
         "   [name] TEXT,\n"
-        "   [species_id] INTEGER,\n"
+        "   [species_id] INTEGER REFERENCES [species]([id])\n"
-        "   FOREIGN KEY([species_id]) REFERENCES [species]([id])\n"
         ")"
     )
     assert list(fresh_db["species"].rows) == [