remote/sqlite-utils
1
0
Fork 0
mirror of https://github.com/simonw/sqlite-utils.git synced 2026-07-23 01:14:31 +02:00
Code Issues Projects Releases Packages Wiki Activity

.add_foreign_keys() uses .transform() instead of PRAGMA writable_schema

Browse source 
Closes #577

This should solve all sorts of problems seen by users of platforms that throw errors on writable_schema.

Also added `add_foreign_keys=` and `foreign_keys=` parameters to `table.transform()`.
This commit is contained in:
Simon Willison 2023-08-17 17:48:08 -07:00 • committed by GitHub
commit 509857ee87
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 336 additions and 143 deletions
Download patch file Download diff file Expand all files Collapse all files

63
docs/python-api.rst
View file

@ -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() <python_api_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 <python_api_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
--------------------------------

132
sqlite_utils/db.py
View file

@ -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]

111
tests/test_cli.py
View file

@ -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)",
),

23
tests/test_create.py
View file

@ -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]) )'
)

19
tests/test_extract.py
View file

@ -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) == [

131
tests/test_transform.py
View file

@ -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"
")"
)