diff --git a/docs/python-api.rst b/docs/python-api.rst index e63c27d..2dd79fe 100644 --- a/docs/python-api.rst +++ b/docs/python-api.rst @@ -1281,9 +1281,7 @@ Adding foreign key constraints The SQLite ``ALTER TABLE`` statement doesn't have the ability to add foreign key references to an existing column. -It's possible to add these references through very careful manipulation of SQLite's ``sqlite_master`` table, using ``PRAGMA writable_schema``. - -``sqlite-utils`` can do this for you, though there is a significant risk of data corruption if something goes wrong so it is advisable to create a fresh copy of your database file before attempting this. +The ``add_foreign_key()`` method here is a convenient wrapper around :ref:`table.transform() `. Here's an example of this mechanism in action: @@ -1317,11 +1315,7 @@ To ignore the case where the key already exists, use ``ignore=True``: Adding multiple foreign key constraints at once ----------------------------------------------- -The final step in adding a new foreign key to a SQLite database is to run ``VACUUM``, to ensure the new foreign key is available in future introspection queries. - -``VACUUM`` against a large (multi-GB) database can take several minutes or longer. If you are adding multiple foreign keys using ``table.add_foreign_key(...)`` these can quickly add up. - -Instead, you can use ``db.add_foreign_keys(...)`` to add multiple foreign keys within a single transaction. This method takes a list of four-tuples, each one specifying a ``table``, ``column``, ``other_table`` and ``other_column``. +You can use ``db.add_foreign_keys(...)`` to add multiple foreign keys in one go. This method takes a list of four-tuples, each one specifying a ``table``, ``column``, ``other_table`` and ``other_column``. Here's an example adding two foreign keys at once: @@ -1388,6 +1382,8 @@ To keep the original table around instead of dropping it, pass the ``keep_table= table.transform(types={"age": int}, keep_table="original_table") +.. _python_api_transform_alter_column_types: + Altering column types --------------------- @@ -1400,6 +1396,8 @@ To alter the type of a column, use the ``types=`` argument: See :ref:`python_api_add_column` for a list of available types. +.. _python_api_transform_rename_columns: + Renaming columns ---------------- @@ -1410,6 +1408,8 @@ The ``rename=`` parameter can rename columns: # Rename 'age' to 'initial_age': table.transform(rename={"age": "initial_age"}) +.. _python_api_transform_drop_columns: + Dropping columns ---------------- @@ -1420,6 +1420,8 @@ To drop columns, pass them in the ``drop=`` set: # Drop the 'age' column: table.transform(drop={"age"}) +.. _python_api_transform_change_primary_keys: + Changing primary keys --------------------- @@ -1430,6 +1432,8 @@ To change the primary key for a table, use ``pk=``. This can be passed a single # Make `user_id` the new primary key table.transform(pk="user_id") +.. _python_api_transform_change_not_null: + Changing not null status ------------------------ @@ -1450,6 +1454,8 @@ If you want to take existing ``NOT NULL`` columns and change them to allow null # Make age allow NULL and switch weight to being NOT NULL: table.transform(not_null={"age": False, "weight": True}) +.. _python_api_transform_alter_column_defaults: + Altering column defaults ------------------------ @@ -1463,6 +1469,8 @@ The ``defaults=`` parameter can be used to set or change the defaults for differ # Now remove the default from that column: table.transform(defaults={"age": None}) +.. _python_api_transform_change_column_order: + Changing column order --------------------- @@ -1473,6 +1481,45 @@ The ``column_order=`` parameter can be used to change the order of the columns. # Change column order table.transform(column_order=("name", "age", "id") +.. _python_api_transform_add_foreign_key_constraints: + +Adding foreign key constraints +------------------------------ + +You can add one or more foreign key constraints to a table using the ``add_foreign_keys=`` parameter: + +.. code-block:: python + + db["places"].transform( + add_foreign_keys=( + ("country", "country", "id"), + ("continent", "continent", "id") + ) + ) + +This accepts the same arguments described in :ref:`specifying foreign keys ` - so you can specify them as a full tuple of ``(column, other_table, other_column)``, or you can take a shortcut and pass just the name of the column, provided the table can be automatically derived from the column name: + +.. code-block:: python + + db["places"].transform( + add_foreign_keys=(("country", "continent")) + ) + +.. _python_api_transform_replace_foreign_key_constraints: + +Replacing foreign key constraints +--------------------------------- + +The ``foreign_keys=`` parameter is similar to to ``add_foreign_keys=`` but can be be used to replace all foreign key constraints on a table, dropping any that are not explicitly mentioned: + +.. code-block:: python + + db["places"].transform( + add_foreign_keys=(("continent",)) + ) + +.. _python_api_transform_drop_foreign_key_constraints: + Dropping foreign key constraints -------------------------------- diff --git a/sqlite_utils/db.py b/sqlite_utils/db.py index 78c0667..de8d1c3 100644 --- a/sqlite_utils/db.py +++ b/sqlite_utils/db.py @@ -156,14 +156,16 @@ XIndexColumn = namedtuple( Trigger = namedtuple("Trigger", ("name", "table", "sql")) -ForeignKeysType = Union[ - Iterable[str], - Iterable[ForeignKey], - Iterable[Tuple[str, str]], - Iterable[Tuple[str, str, str]], - Iterable[Tuple[str, str, str, str]], +ForeignKeyIndicator = Union[ + str, + ForeignKey, + Tuple[str, str], + Tuple[str, str, str], + Tuple[str, str, str, str], ] +ForeignKeysType = Union[Iterable[ForeignKeyIndicator], List[ForeignKeyIndicator]] + class Default: pass @@ -747,9 +749,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 - # it into a list of ForeignKey tuples + """ + Given a list of differing foreign_keys definitions, return a list of + 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,12 +776,20 @@ 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) assert len(tuple_or_list) in ( 2, 3, + 4, ), "foreign_keys= should be a list of tuple pairs or triples" - if len(tuple_or_list) == 3: - tuple_or_list = cast(Tuple[str, str, str], tuple_or_list) + if len(tuple_or_list) in (3, 4): + if len(tuple_or_list) == 4: + tuple_or_list = cast(Tuple[str, str, str], tuple_or_list[1:]) + else: + tuple_or_list = cast(Tuple[str, str, str], tuple_or_list) fks.append( ForeignKey( name, tuple_or_list[0], tuple_or_list[1], tuple_or_list[2] @@ -864,7 +881,7 @@ class Database: for fk in foreign_keys: if fk.other_table == name and columns.get(fk.other_column): continue - if not any( + if fk.other_column != "rowid" and not any( c for c in self[fk.other_table].columns if c.name == fk.other_column ): raise AlterError( @@ -1148,32 +1165,14 @@ class Database: (table, column, other_table, other_column) ) - # 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 + # Group them by table + by_table: Dict[str, List] = {} + for fk in foreign_keys_to_create: + by_table.setdefault(fk[0], []).append(fk) + + for table, fks in by_table.items(): + cast(Table, self[table]).transform(add_foreign_keys=fks) - # 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 +1703,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 +1722,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 +1738,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 +1770,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 +1786,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 +1797,45 @@ class Table(Queryable): types = types or {} rename = rename or {} drop = drop or set() + + create_table_foreign_keys: List[ForeignKeyIndicator] = [] + + 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.extend(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 +1895,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] diff --git a/tests/test_cli.py b/tests/test_cli.py index 305535b..d255abc 100644 --- a/tests/test_cli.py +++ b/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 ( - "CREATE TABLE [books] ( [title] TEXT , [author_id] INTEGER, FOREIGN KEY([author_id]) REFERENCES [authors]([id]) )" - == collapse_whitespace(db["books"].schema) + assert result.exit_code == 0, result.output + assert db["books"].schema == ( + 'CREATE TABLE "books" (\n' + " [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 ( - "CREATE TABLE [books] ( [title] TEXT , [author_id] INTEGER, [author_name_ref] TEXT, " - "FOREIGN KEY([author_id]) REFERENCES [authors]([id]), " - "FOREIGN KEY([author_name_ref]) REFERENCES [authors]([name]) )" - == collapse_whitespace(db["books"].schema) + assert result.exit_code == 0, result.output + assert db["books"].schema == ( + 'CREATE TABLE "books" (\n' + " [title] TEXT,\n" + " [author_id] INTEGER REFERENCES [authors]([id]),\n" + " [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" - " FOREIGN KEY([species_id]) REFERENCES [species]([id])\n" + " [species_id] INTEGER 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" - " FOREIGN KEY([custom_table_id]) REFERENCES [custom_table]([id])\n" + " [custom_table_id] INTEGER 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" - " FOREIGN KEY([custom_fk]) REFERENCES [species]([id])\n" + " [custom_fk] INTEGER 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" - " FOREIGN KEY([species_id]) REFERENCES [species]([id])\n" + " [species_id] INTEGER REFERENCES [species]([id])\n" ")", "CREATE TABLE [species] (\n [id] INTEGER PRIMARY KEY,\n [species] TEXT\n)", ), diff --git a/tests/test_create.py b/tests/test_create.py index 63f0991..9deb2a5 100644 --- a/tests/test_create.py +++ b/tests/test_create.py @@ -469,18 +469,21 @@ def test_add_column_foreign_key(fresh_db): fresh_db.create_table("dogs", {"name": str}) fresh_db.create_table("breeds", {"name": str}) fresh_db["dogs"].add_column("breed_id", fk="breeds") - assert ( - "CREATE TABLE [dogs] ( [name] TEXT , [breed_id] INTEGER, FOREIGN KEY([breed_id]) REFERENCES [breeds]([rowid]) )" - == collapse_whitespace(fresh_db["dogs"].schema) + assert fresh_db["dogs"].schema == ( + 'CREATE TABLE "dogs" (\n' + " [name] TEXT,\n" + " [breed_id] INTEGER REFERENCES [breeds]([rowid])\n" + ")" ) # And again with an explicit primary key column fresh_db.create_table("subbreeds", {"name": str, "primkey": str}, pk="primkey") fresh_db["dogs"].add_column("subbreed_id", fk="subbreeds") - assert ( - "CREATE TABLE [dogs] ( [name] TEXT , [breed_id] INTEGER, [subbreed_id] TEXT, " - "FOREIGN KEY([breed_id]) REFERENCES [breeds]([rowid]), " - "FOREIGN KEY([subbreed_id]) REFERENCES [subbreeds]([primkey]) )" - == collapse_whitespace(fresh_db["dogs"].schema) + assert fresh_db["dogs"].schema == ( + 'CREATE TABLE "dogs" (\n' + " [name] TEXT,\n" + " [breed_id] INTEGER REFERENCES [breeds]([rowid]),\n" + " [subbreed_id] TEXT REFERENCES [subbreeds]([primkey])\n" + ")" ) @@ -490,8 +493,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]) )' ) diff --git a/tests/test_extract.py b/tests/test_extract.py index 70ad0cf..7a663c5 100644 --- a/tests/test_extract.py +++ b/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) - + " [end] INTEGER,\n" - + " FOREIGN KEY([{}]) REFERENCES [{}]([id])\n".format( - expected_fk, expected_table - ) + " [{}] INTEGER REFERENCES [{}]([id]),\n".format(expected_fk, expected_table) + + " [end] INTEGER\n" + ")" ) 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" - " 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 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" - " 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 ( @@ -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" - " FOREIGN KEY([species_id]) REFERENCES [species]([id])\n" + " [species_id] INTEGER 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" - " FOREIGN KEY([species_id]) REFERENCES [species]([id])\n" + " [species_id] INTEGER REFERENCES [species]([id])\n" ")" ) assert list(fresh_db["species"].rows) == [ diff --git a/tests/test_transform.py b/tests/test_transform.py index 6b0b53a..1136f7c 100644 --- a/tests/test_transform.py +++ b/tests/test_transform.py @@ -329,22 +329,29 @@ def test_transform_foreign_keys_survive_renamed_column( ] +def _add_country_city_continent(db): + db["country"].insert({"id": 1, "name": "France"}, pk="id") + db["continent"].insert({"id": 2, "name": "Europe"}, pk="id") + db["city"].insert({"id": 24, "name": "Paris"}, pk="id") + + +_CAVEAU = { + "id": 32, + "name": "Caveau de la Huchette", + "country": 1, + "continent": 2, + "city": 24, +} + + @pytest.mark.parametrize("use_pragma_foreign_keys", [False, True]) def test_transform_drop_foreign_keys(fresh_db, use_pragma_foreign_keys): if use_pragma_foreign_keys: fresh_db.conn.execute("PRAGMA foreign_keys=ON") # Create table with three foreign keys so we can drop two of them - fresh_db["country"].insert({"id": 1, "name": "France"}, pk="id") - fresh_db["continent"].insert({"id": 2, "name": "Europe"}, pk="id") - fresh_db["city"].insert({"id": 24, "name": "Paris"}, pk="id") + _add_country_city_continent(fresh_db) fresh_db["places"].insert( - { - "id": 32, - "name": "Caveau de la Huchette", - "country": 1, - "continent": 2, - "city": 24, - }, + _CAVEAU, foreign_keys=("country", "continent", "city"), ) assert fresh_db["places"].foreign_keys == [ @@ -387,3 +394,107 @@ def test_transform_verify_foreign_keys(fresh_db): == "CREATE TABLE [authors] (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT\n)" ) assert fresh_db.conn.execute("PRAGMA foreign_keys").fetchone()[0] + + +def test_transform_add_foreign_keys_from_scratch(fresh_db): + _add_country_city_continent(fresh_db) + fresh_db["places"].insert(_CAVEAU) + # Should have no foreign keys + assert fresh_db["places"].foreign_keys == [] + # Now add them using .transform() + fresh_db["places"].transform(add_foreign_keys=("country", "continent", "city")) + # Should now have all three: + assert fresh_db["places"].foreign_keys == [ + ForeignKey( + table="places", column="city", other_table="city", other_column="id" + ), + ForeignKey( + table="places", + column="continent", + other_table="continent", + other_column="id", + ), + ForeignKey( + table="places", column="country", other_table="country", other_column="id" + ), + ] + assert fresh_db["places"].schema == ( + 'CREATE TABLE "places" (\n' + " [id] INTEGER,\n" + " [name] TEXT,\n" + " [country] INTEGER REFERENCES [country]([id]),\n" + " [continent] INTEGER REFERENCES [continent]([id]),\n" + " [city] INTEGER REFERENCES [city]([id])\n" + ")" + ) + + +@pytest.mark.parametrize( + "add_foreign_keys", + ( + ("country", "continent"), + # Fully specified + ( + ("country", "country", "id"), + ("continent", "continent", "id"), + ), + ), +) +def test_transform_add_foreign_keys_from_partial(fresh_db, add_foreign_keys): + _add_country_city_continent(fresh_db) + fresh_db["places"].insert( + _CAVEAU, + foreign_keys=("city",), + ) + # Should have one foreign keys + assert fresh_db["places"].foreign_keys == [ + ForeignKey(table="places", column="city", other_table="city", other_column="id") + ] + # Now add three more using .transform() + fresh_db["places"].transform(add_foreign_keys=add_foreign_keys) + # Should now have all three: + assert fresh_db["places"].foreign_keys == [ + ForeignKey( + table="places", column="city", other_table="city", other_column="id" + ), + ForeignKey( + table="places", + column="continent", + other_table="continent", + other_column="id", + ), + ForeignKey( + table="places", column="country", other_table="country", other_column="id" + ), + ] + + +@pytest.mark.parametrize( + "foreign_keys", + ( + ("country", "continent"), + # Fully specified + ( + ("country", "country", "id"), + ("continent", "continent", "id"), + ), + ), +) +def test_transform_replace_foreign_keys(fresh_db, foreign_keys): + _add_country_city_continent(fresh_db) + fresh_db["places"].insert( + _CAVEAU, + foreign_keys=("city",), + ) + assert len(fresh_db["places"].foreign_keys) == 1 + # Replace with two different ones + fresh_db["places"].transform(foreign_keys=foreign_keys) + assert fresh_db["places"].schema == ( + 'CREATE TABLE "places" (\n' + " [id] INTEGER,\n" + " [name] TEXT,\n" + " [country] INTEGER REFERENCES [country]([id]),\n" + " [continent] INTEGER REFERENCES [continent]([id]),\n" + " [city] INTEGER\n" + ")" + )