diff --git a/Justfile b/Justfile index f4f0e7f..5caa120 100644 --- a/Justfile +++ b/Justfile @@ -8,11 +8,12 @@ @run *options: uv run -- {{options}} -# Run linters: black, flake8, mypy, cog +# Run linters: black, flake8, mypy, ty, cog @lint: just run black . --check uv run flake8 uv run mypy sqlite_utils tests + uv run ty check sqlite_utils uv run cog --check README.md docs/*.rst uv run --group docs codespell docs/*.rst --ignore-words docs/codespell-ignore-words.txt diff --git a/docs/changelog.rst b/docs/changelog.rst index 2e1f69c..f6a8ea2 100644 --- a/docs/changelog.rst +++ b/docs/changelog.rst @@ -4,6 +4,29 @@ Changelog =========== +.. _v_unreleased: + +Unreleased +---------- + +Breaking changes: + +- ``table.foreign_keys`` now returns ``ForeignKey`` objects that are dataclasses rather than ``namedtuple`` instances, so they can no longer be unpacked or indexed as ``(table, column, other_table, other_column)`` tuples - access their fields by name instead. Compound (multi-column) foreign keys are now represented as a single ``ForeignKey`` with ``is_compound=True`` and populated ``columns``/``other_columns`` tuples, where ``column`` and ``other_column`` are ``None``. Previously they were returned as one ``ForeignKey`` per column, misleadingly suggesting several independent foreign keys. See :ref:`upgrading_3_to_4` for details. (:issue:`594`) + +Compound foreign key support: + +- Tables can now be created with compound foreign keys, by passing tuples of column names in ``foreign_keys=``: ``foreign_keys=[(("campus_name", "dept_code"), "departments")]``. The referenced columns default to the compound primary key of the other table. Compound keys are rendered as table-level ``FOREIGN KEY`` constraints in the generated schema. See :ref:`python_api_compound_foreign_keys`. +- ``table.transform()`` now preserves compound foreign keys, applying any column renames to them. Dropping a column that is part of a compound foreign key drops the whole constraint, matching the existing single-column behavior. ``drop_foreign_keys=`` accepts a bare column name - dropping any foreign key that column participates in - or a tuple of columns to target a compound key precisely. +- ``table.add_foreign_key()`` and ``db.add_foreign_keys()`` accept tuples of column names to add a compound foreign key to an existing table. +- ``db.index_foreign_keys()`` creates a single composite index for a compound foreign key. + +Other foreign key improvements: + +- ``ForeignKey`` now exposes ``on_delete`` and ``on_update`` fields reflecting the foreign key's ``ON DELETE``/``ON UPDATE`` actions, and ``table.transform()`` preserves those actions. Previously a transform silently stripped clauses such as ``ON DELETE CASCADE`` from the table schema. +- ``table.add_foreign_key()`` accepts new ``on_delete=`` and ``on_update=`` parameters for creating foreign keys with actions, e.g. ``table.add_foreign_key("author_id", "authors", "id", on_delete="CASCADE")``. (:issue:`530`) +- Foreign keys declared as ``REFERENCES other_table`` with no explicit column are now resolved to the other table's primary key by ``table.foreign_keys``, instead of reporting ``other_column=None``. +- Fixed a ``TypeError`` when sorting ``ForeignKey`` objects where some were compound. + .. _v4_0rc2: 4.0rc2 (2026-07-04) diff --git a/docs/python-api.rst b/docs/python-api.rst index 3f7e12c..a33b256 100644 --- a/docs/python-api.rst +++ b/docs/python-api.rst @@ -817,6 +817,61 @@ You can leave off the third item in the tuple to have the referenced column auto ("author_id", "authors") ]) +.. _python_api_compound_foreign_keys: + +Compound foreign keys +~~~~~~~~~~~~~~~~~~~~~ + +To create a compound (multi-column) foreign key, use tuples of column names in place of the single column names: + +.. code-block:: python + + db.table("courses").create({ + "course_code": str, + "campus_name": str, + "dept_code": str, + }, pk="course_code", foreign_keys=[ + (("campus_name", "dept_code"), "departments", ("campus_name", "dept_code")) + ]) + +This creates a table-level constraint: + +.. code-block:: sql + + CREATE TABLE "courses" ( + "course_code" TEXT PRIMARY KEY, + "campus_name" TEXT, + "dept_code" TEXT, + FOREIGN KEY ("campus_name", "dept_code") REFERENCES "departments"("campus_name", "dept_code") + ) + +As with single columns, you can leave off the tuple of other columns to reference the compound primary key of the other table: + +.. code-block:: python + + foreign_keys=[ + (("campus_name", "dept_code"), "departments") + ] + +To specify ``ON DELETE`` or ``ON UPDATE`` actions, pass ``ForeignKey`` objects instead: + +.. code-block:: python + + from sqlite_utils.db import ForeignKey + + db.table("books").create({ + "id": int, + "author_id": int, + }, pk="id", foreign_keys=[ + ForeignKey( + table="books", column="author_id", + other_table="authors", other_column="id", + on_delete="CASCADE", + ) + ]) + +Foreign key actions are preserved by :ref:`table.transform() ` - prior to sqlite-utils 4.0 they were silently dropped when a table was transformed. + .. _python_api_table_configuration: Table configuration options @@ -1526,6 +1581,26 @@ To ignore the case where the key already exists, use ``ignore=True``: db.table("books").add_foreign_key("author_id", "authors", "id", ignore=True) +To add a compound foreign key, pass tuples of columns: + +.. code-block:: python + + db.table("courses").add_foreign_key( + ("campus_name", "dept_code"), "departments", ("campus_name", "dept_code") + ) + +As with single columns, omitting the other columns will use the compound primary key of the other table. ``other_table`` must always be specified for a compound foreign key. + +Use ``on_delete=`` and ``on_update=`` to specify ``ON DELETE`` and ``ON UPDATE`` actions for the foreign key: + +.. code-block:: python + + db.table("books").add_foreign_key( + "author_id", "authors", "id", on_delete="CASCADE" + ) + +This creates a foreign key with an ``ON DELETE CASCADE`` clause, so deleting an author will also delete their books (provided foreign key enforcement is enabled with ``PRAGMA foreign_keys = ON``). Valid actions are ``"SET NULL"``, ``"SET DEFAULT"``, ``"CASCADE"``, ``"RESTRICT"`` and the default ``"NO ACTION"``. + .. _python_api_add_foreign_keys: Adding multiple foreign key constraints at once @@ -1555,6 +1630,8 @@ If you want to ensure that every foreign key column in your database has a corre db.index_foreign_keys() +Compound foreign keys get a single composite index across their columns. + .. _python_api_drop: Dropping a table or view @@ -1757,6 +1834,16 @@ This example drops two foreign keys - the one from ``places.country`` to ``count drop_foreign_keys=("country", "continent") ) +A bare column name drops any foreign key that column participates in, including compound foreign keys. To target a compound foreign key precisely, pass a tuple of its columns: + +.. code-block:: python + + db.table("courses").transform( + drop_foreign_keys=[("campus_name", "dept_code")] + ) + +Renaming a column with ``rename=`` updates any foreign keys that use it, and dropping a column with ``drop=`` also drops any foreign keys it participates in - for a compound foreign key this removes the whole constraint. + .. _python_api_transform_sql: Custom transformations with .transform_sql() @@ -2204,17 +2291,39 @@ Almost all SQLite tables have a ``rowid`` column, but a table with no explicitly .foreign_keys ------------- -The ``.foreign_keys`` property returns any foreign key relationships for the table, as a list of ``ForeignKey(table, column, other_table, other_column)`` named tuples. It is not available on views. +The ``.foreign_keys`` property returns any foreign key relationships for the table, as a list of ``ForeignKey`` objects. It is not available on views. + +Each ``ForeignKey`` has the following attributes: + +``table`` + The table the foreign key is defined on. +``column`` + The column on this table, or ``None`` for a compound foreign key. +``other_table`` + The table being referenced. +``other_column`` + The referenced column, or ``None`` for a compound foreign key. +``columns`` + A tuple of the columns on this table, always populated (a one-item tuple for single-column foreign keys). +``other_columns`` + A tuple of the referenced columns. +``is_compound`` + ``True`` if this is a compound (multi-column) foreign key. +``on_delete`` + The ``ON DELETE`` action, e.g. ``"CASCADE"`` - ``"NO ACTION"`` if not set. +``on_update`` + The ``ON UPDATE`` action - ``"NO ACTION"`` if not set. + +``ForeignKey`` was a ``namedtuple`` prior to sqlite-utils 4.0. It is now a dataclass and can no longer be unpacked or indexed as a tuple - access its fields by name instead. See :ref:`upgrading_3_to_4` for details. :: >>> db.table("Street_Tree_List").foreign_keys - [ForeignKey(table='Street_Tree_List', column='qLegalStatus', other_table='qLegalStatus', other_column='id'), - ForeignKey(table='Street_Tree_List', column='qCareAssistant', other_table='qCareAssistant', other_column='id'), - ForeignKey(table='Street_Tree_List', column='qSiteInfo', other_table='qSiteInfo', other_column='id'), - ForeignKey(table='Street_Tree_List', column='qSpecies', other_table='qSpecies', other_column='id'), - ForeignKey(table='Street_Tree_List', column='qCaretaker', other_table='qCaretaker', other_column='id'), - ForeignKey(table='Street_Tree_List', column='PlantType', other_table='PlantType', other_column='id')] + [ForeignKey(table='Street_Tree_List', column='qLegalStatus', other_table='qLegalStatus', other_column='id', columns=('qLegalStatus',), other_columns=('id',), is_compound=False, on_delete='NO ACTION', on_update='NO ACTION'), + ForeignKey(table='Street_Tree_List', column='qCareAssistant', other_table='qCareAssistant', other_column='id', columns=('qCareAssistant',), other_columns=('id',), is_compound=False, on_delete='NO ACTION', on_update='NO ACTION'), + ...] + +Compound foreign keys - defined with ``FOREIGN KEY (col_a, col_b) REFERENCES other(col_a, col_b)`` - are returned as a single ``ForeignKey`` with ``is_compound=True``, ``column`` and ``other_column`` set to ``None``, and the participating columns available in the ``columns`` and ``other_columns`` tuples. .. _python_api_introspection_schema: diff --git a/docs/reference.rst b/docs/reference.rst index 5b5fd25..a9fdf29 100644 --- a/docs/reference.rst +++ b/docs/reference.rst @@ -70,6 +70,13 @@ sqlite_utils.db.ColumnDetails .. autoclass:: sqlite_utils.db.ColumnDetails +.. _reference_db_other_foreign_key: + +sqlite_utils.db.ForeignKey +-------------------------- + +.. autoclass:: sqlite_utils.db.ForeignKey + sqlite_utils.utils ================== diff --git a/docs/upgrading.rst b/docs/upgrading.rst index f743a3e..e7fd273 100644 --- a/docs/upgrading.rst +++ b/docs/upgrading.rst @@ -79,6 +79,24 @@ Python API changes **View.enable_fts() has been removed.** The ``View`` class previously had an ``enable_fts()`` method that existed only to raise ``NotImplementedError`` - full-text search is not supported for views. Calling it now raises ``AttributeError`` like any other missing method. +**ForeignKey is now a dataclass, not a namedtuple.** The ``ForeignKey`` objects returned by ``table.foreign_keys`` gained new fields - ``columns``, ``other_columns``, ``is_compound``, ``on_delete`` and ``on_update`` - so that compound (multi-column) foreign keys and foreign key actions can be represented. To make room for those fields cleanly ``ForeignKey`` is now a dataclass rather than a ``namedtuple``, so it can no longer be unpacked or indexed as a tuple. Access its fields by name instead: + +.. code-block:: python + + # 3.x - tuple unpacking, no longer works: + for table, column, other_table, other_column in db["courses"].foreign_keys: + ... + + # 4.0 - access fields by name: + for fk in db["courses"].foreign_keys: + fk.table, fk.column, fk.other_table, fk.other_column + +Attempting the old unpacking or ``fk[0]`` indexing now raises ``TypeError``, so any code using those patterns will fail loudly rather than silently misbehave. + +Compound foreign keys - previously returned as one ``ForeignKey`` per column, misleadingly suggesting several independent single-column keys - are now returned as a single ``ForeignKey`` with ``is_compound=True``. For these the scalar ``column`` and ``other_column`` fields are ``None``; use the ``columns`` and ``other_columns`` tuples instead. Single-column foreign keys are unaffected apart from the class change: ``column``/``other_column`` behave as before and ``columns``/``other_columns`` are one-item tuples. + +Two related behavior changes to ``table.transform()``: compound foreign keys now survive a transform (previously they were split into separate single-column keys), and ``ON DELETE``/``ON UPDATE`` actions such as ``ON DELETE CASCADE`` are now preserved (previously they were silently stripped from the schema). + **Validation errors raise ValueError.** Invalid arguments to Python API methods - for example ``create_table()`` with no columns, or ``ignore=True`` together with ``replace=True`` - now raise ``ValueError``. They previously raised ``AssertionError`` from bare ``assert`` statements, which were silently skipped under ``python -O``. **Transaction behavior is now well-defined.** 4.0 introduces the :ref:`db.atomic() ` context manager and uses it consistently for every write operation - the full model is described in :ref:`python_api_transactions`. Changes you may notice: diff --git a/sqlite_utils/db.py b/sqlite_utils/db.py index a8f8e17..3f23c7d 100644 --- a/sqlite_utils/db.py +++ b/sqlite_utils/db.py @@ -11,6 +11,7 @@ from .utils import ( ) import binascii from collections import namedtuple +from dataclasses import dataclass, field from collections.abc import Mapping import contextlib import datetime @@ -161,9 +162,65 @@ Summary information about a column, see :ref:`python_api_analyze_column`. The ``N`` least common values as a list of ``(value, count)`` tuples, or ``None`` if the table is entirely distinct or if the number of distinct values is less than N (since they will already have been returned in ``most_common``) """ -ForeignKey = namedtuple( - "ForeignKey", ("table", "column", "other_table", "other_column") -) + + +@dataclass(order=True) +class ForeignKey: + """ + A foreign key defined on a table. + + For single-column foreign keys ``column`` and ``other_column`` hold the + column names, and ``columns``/``other_columns`` are one-item tuples. + + For compound (multi-column) foreign keys ``column`` and ``other_column`` + are ``None`` - use ``columns`` and ``other_columns`` instead, and check + ``is_compound``. + + ``on_delete`` and ``on_update`` hold the foreign key actions, e.g. + ``"CASCADE"`` - ``"NO ACTION"`` if not set. + + Prior to sqlite-utils 4.0 this was a ``namedtuple`` and could be unpacked + or indexed as ``(table, column, other_table, other_column)``. It is now a + dataclass - access its fields by name instead. + """ + + table: str + # column/other_column are None for compound keys, which would break + # ordering against str values - comparison uses columns/other_columns + column: Optional[str] = field(compare=False) + other_table: str + other_column: Optional[str] = field(compare=False) + columns: Tuple[str, ...] = () + other_columns: Tuple[str, ...] = () + is_compound: bool = False + on_delete: str = "NO ACTION" + on_update: str = "NO ACTION" + + def __post_init__(self): + # Populate columns/other_columns for single-column foreign keys, + # normalizing any lists to tuples + if self.columns: + self.columns = tuple(self.columns) + else: + self.columns = (self.column,) if self.column is not None else () + if self.other_columns: + self.other_columns = tuple(self.other_columns) + else: + self.other_columns = ( + (self.other_column,) if self.other_column is not None else () + ) + + +def _fk_actions_sql(fk: ForeignKey) -> str: + "ON UPDATE/ON DELETE clauses for a foreign key, or an empty string." + actions = "" + if fk.on_update and fk.on_update != "NO ACTION": + actions += " ON UPDATE {}".format(fk.on_update) + if fk.on_delete and fk.on_delete != "NO ACTION": + actions += " ON DELETE {}".format(fk.on_delete) + return actions + + Index = namedtuple("Index", ("seq", "name", "unique", "origin", "partial", "columns")) XIndex = namedtuple("XIndex", ("name", "columns")) XIndexColumn = namedtuple( @@ -176,12 +233,18 @@ class TransformError(Exception): pass +# A single column name, or a tuple of columns for a compound foreign key +ForeignKeyColumns = Union[str, Tuple[str, ...], List[str]] + +# (table, column(s), other_table, other_column(s)) +ForeignKeyTuple = Tuple[str, ForeignKeyColumns, str, ForeignKeyColumns] + ForeignKeyIndicator = Union[ str, ForeignKey, - Tuple[str, str], - Tuple[str, str, str], - Tuple[str, str, str, str], + Tuple[ForeignKeyColumns, str], + Tuple[ForeignKeyColumns, str, ForeignKeyColumns], + ForeignKeyTuple, ] ForeignKeysType = Union[Iterable[ForeignKeyIndicator], List[ForeignKeyIndicator]] @@ -1105,9 +1168,12 @@ class Database: :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), + string, a ForeignKey() object, 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, column, other_table, other_column). For compound foreign + keys the column elements can be tuples of column names, e.g. + (("campus_name", "dept_code"), "departments") or + (("campus_name", "dept_code"), "departments", ("campus_name", "dept_code")) """ table = self.table(name) if all(isinstance(fk, ForeignKey) for fk in foreign_keys): @@ -1124,7 +1190,7 @@ class Database: if not all(isinstance(fk, (tuple, list)) for fk in foreign_keys): raise ValueError("foreign_keys= should be a list of tuples") fks = [] - for tuple_or_list in foreign_keys: + for tuple_or_list in cast(Iterable[Sequence[Any]], foreign_keys): if len(tuple_or_list) == 4: if tuple_or_list[0] != name: raise ValueError( @@ -1132,28 +1198,60 @@ class Database: tuple_or_list, name ) ) - if len(tuple_or_list) not in (2, 3, 4): + tuple_or_list = tuple_or_list[1:] + if len(tuple_or_list) not in (2, 3): raise ValueError( "foreign_keys= should be a list of tuple pairs or triples" ) - 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:]) + column_or_columns = tuple_or_list[0] + other_table = tuple_or_list[1] + if isinstance(column_or_columns, (list, tuple)): + # Compound foreign key + columns = tuple(column_or_columns) + if len(tuple_or_list) == 3: + if not isinstance(tuple_or_list[2], (list, tuple)): + raise ValueError( + "Compound foreign key {} should reference a tuple " + "of other columns".format(tuple(tuple_or_list)) + ) + other_columns = tuple(tuple_or_list[2]) 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] + # Guess the compound primary key of the other table + other_columns = tuple(self.table(other_table).pks) + if len(columns) != len(other_columns): + raise ValueError( + "Compound foreign key {} should have the same number " + "of columns on both sides".format(tuple(tuple_or_list)) ) + if len(columns) == 1: + # Single-column key passed as a one-item list + fks.append( + ForeignKey(name, columns[0], other_table, other_columns[0]) + ) + else: + fks.append( + ForeignKey( + name, + None, + other_table, + None, + columns=columns, + other_columns=other_columns, + is_compound=True, + ) + ) + elif len(tuple_or_list) == 3: + fks.append( + ForeignKey(name, column_or_columns, other_table, tuple_or_list[2]) ) else: # Guess the primary key fks.append( ForeignKey( name, - tuple_or_list[0], - tuple_or_list[1], - table.guess_foreign_column(tuple_or_list[1]), + column_or_columns, + other_table, + table.guess_foreign_column(other_table), ) ) return fks @@ -1192,7 +1290,11 @@ class Database: if hash_id_columns and (hash_id is None): hash_id = "id" foreign_keys = self.resolve_foreign_keys(name, foreign_keys or []) - foreign_keys_by_column = {fk.column: fk for fk in foreign_keys} + # Compound foreign keys are rendered as table-level constraints; + # single-column ones as inline REFERENCES on their column + foreign_keys_by_column = { + fk.column: fk for fk in foreign_keys if not fk.is_compound + } # any extracts will be treated as integer columns with a foreign key extracts = resolve_extracts(extracts) for extract_column, extract_table in extracts.items(): @@ -1234,14 +1336,15 @@ class Database: pk = hash_id # Soundness check foreign_keys point to existing tables for fk in foreign_keys: - if fk.other_table == name and columns.get(fk.other_column): - continue - 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( - "No such column: {}.{}".format(fk.other_table, fk.other_column) - ) + for other_column in fk.other_columns: + if fk.other_table == name and columns.get(other_column): + continue + if other_column != "rowid" and not any( + c for c in self[fk.other_table].columns if c.name == other_column + ): + raise AlterError( + "No such column: {}.{}".format(fk.other_table, other_column) + ) column_defs = [] # ensure pk is a tuple @@ -1263,14 +1366,12 @@ class Database: "DEFAULT {}".format(self.quote_default_value(defaults[column_name])) ) if column_name in foreign_keys_by_column: + fk = foreign_keys_by_column[column_name] column_extras.append( - "REFERENCES {}({})".format( - quote_identifier( - foreign_keys_by_column[column_name].other_table - ), - quote_identifier( - foreign_keys_by_column[column_name].other_column - ), + "REFERENCES {}({}){}".format( + quote_identifier(fk.other_table), + quote_identifier(cast(str, fk.other_column)), + _fk_actions_sql(fk), ) ) column_type_str = COLUMN_TYPE_MAPPING[column_type] @@ -1292,6 +1393,26 @@ class Database: extra_pk = ",\n PRIMARY KEY ({pks})".format( pks=", ".join([quote_identifier(p) for p in pk]) ) + # Compound foreign keys become table-level FOREIGN KEY constraints + column_names = [c[0] for c in column_items] + for fk in foreign_keys: + if not fk.is_compound: + continue + missing = [c for c in fk.columns if c not in column_names] + if missing: + raise AlterError( + "No such column: {}".format(", ".join(sorted(missing))) + ) + column_defs.append( + " FOREIGN KEY ({columns}) REFERENCES {other_table}({other_columns}){actions}".format( + columns=", ".join(quote_identifier(c) for c in fk.columns), + other_table=quote_identifier(fk.other_table), + other_columns=", ".join( + quote_identifier(c) for c in fk.other_columns + ), + actions=_fk_actions_sql(fk), + ) + ) columns_sql = ",\n".join(column_defs) sql = """CREATE TABLE {if_not_exists}{table} ( {columns_sql}{extra_pk} @@ -1489,57 +1610,95 @@ class Database: return candidates def add_foreign_keys( - self, foreign_keys: Iterable[Tuple[str, str, str, str]] + self, foreign_keys: Iterable[Union[ForeignKey, ForeignKeyTuple]] ) -> None: """ See :ref:`python_api_add_foreign_keys`. :param foreign_keys: A list of ``(table, column, other_table, other_column)`` - tuples + tuples - for compound foreign keys, ``column`` and ``other_column`` can + be tuples of column names """ # foreign_keys is a list of explicit 4-tuples if not all( - len(fk) == 4 and isinstance(fk, (list, tuple)) for fk in foreign_keys + isinstance(fk, ForeignKey) + or (isinstance(fk, (list, tuple)) and len(fk) == 4) + for fk in foreign_keys ): raise ValueError( "foreign_keys must be a list of 4-tuples, " "(table, column, other_table, other_column)" ) - foreign_keys_to_create = [] + foreign_keys_to_create: List[ForeignKey] = [] # Verify that all tables and columns exist - for table, column, other_table, other_column in foreign_keys: + for fk in foreign_keys: + if isinstance(fk, ForeignKey): + fk_object = fk + else: + table, column_or_columns, other_table, other_column_or_columns = fk + # Compound foreign keys use tuples of columns + columns = ( + (column_or_columns,) + if isinstance(column_or_columns, str) + else tuple(column_or_columns) + ) + other_columns = ( + (other_column_or_columns,) + if isinstance(other_column_or_columns, str) + else tuple(other_column_or_columns) + ) + if len(columns) == 1: + fk_object = ForeignKey( + table, columns[0], other_table, other_columns[0] + ) + else: + fk_object = ForeignKey( + table, + None, + other_table, + None, + columns=columns, + other_columns=other_columns, + is_compound=True, + ) + table = fk_object.table + columns = fk_object.columns + other_table = fk_object.other_table + other_columns = fk_object.other_columns if not self.table(table).exists(): raise AlterError("No such table: {}".format(table)) table_obj = self.table(table) - if column not in table_obj.columns_dict: - raise AlterError("No such column: {} in {}".format(column, table)) + for column in columns: + if column not in table_obj.columns_dict: + raise AlterError("No such column: {} in {}".format(column, table)) if not self[other_table].exists(): raise AlterError("No such other_table: {}".format(other_table)) - if ( - other_column != "rowid" - and other_column not in self[other_table].columns_dict - ): - raise AlterError( - "No such other_column: {} in {}".format(other_column, other_table) - ) + for other_column in other_columns: + if ( + other_column != "rowid" + and other_column not in self[other_table].columns_dict + ): + raise AlterError( + "No such other_column: {} in {}".format( + other_column, other_table + ) + ) # We will silently skip foreign keys that exist already if not any( fk for fk in table_obj.foreign_keys - if fk.column == column + if fk.columns == columns and fk.other_table == other_table - and fk.other_column == other_column + and fk.other_columns == other_columns ): - foreign_keys_to_create.append( - (table, column, other_table, other_column) - ) + foreign_keys_to_create.append(fk_object) # Group them by table - by_table: Dict[str, List] = {} - for fk in foreign_keys_to_create: - by_table.setdefault(fk[0], []).append(fk) + by_table: Dict[str, List[ForeignKey]] = {} + for fk_object in foreign_keys_to_create: + by_table.setdefault(fk_object.table, []).append(fk_object) for table, fks in by_table.items(): self.table(table).transform(add_foreign_keys=fks) @@ -1551,12 +1710,12 @@ class Database: "Create indexes for every foreign key column on every table in the database." for table_name in self.table_names(): table = self.table(table_name) - existing_indexes = { - i.columns[0] for i in table.indexes if len(i.columns) == 1 - } + existing_indexes = {tuple(i.columns) for i in table.indexes} for fk in table.foreign_keys: - if fk.column not in existing_indexes: - table.create_index([fk.column], find_unique_name=True) + # A compound foreign key gets a single composite index + if fk.columns not in existing_indexes: + table.create_index(fk.columns, find_unique_name=True) + existing_indexes.add(fk.columns) def vacuum(self) -> None: "Run a SQLite ``VACUUM`` against the database." @@ -1907,21 +2066,50 @@ class Table(Queryable): @property def foreign_keys(self) -> List["ForeignKey"]: - "List of foreign keys defined on this table." - fks = [] + """ + List of foreign keys defined on this table. + + Compound (multi-column) foreign keys are returned as a single + ``ForeignKey`` with ``is_compound=True`` and populated + ``columns``/``other_columns`` lists. + """ + # PRAGMA foreign_key_list returns one row per column, grouped by "id" + # with "seq" giving the column order within a compound foreign key. + by_id: Dict[int, list] = {} for row in self.db.execute( "PRAGMA foreign_key_list({})".format(quote_identifier(self.name)) ).fetchall(): if row is not None: id, seq, table_name, from_, to_, on_update, on_delete, match = row - fks.append( - ForeignKey( - table=self.name, - column=from_, - other_table=table_name, - other_column=to_, - ) + by_id.setdefault(id, []).append( + (seq, table_name, from_, to_, on_update, on_delete) ) + fks = [] + for id in sorted(by_id): + rows = sorted(by_id[id]) # order columns by seq + other_table = rows[0][1] + columns = tuple(row[2] for row in rows) + other_columns = tuple(row[3] for row in rows) + if all(c is None for c in other_columns): + # "REFERENCES other_table" with no columns - the pragma + # returns None, meaning the other table's primary key + other_table_pks = tuple(self.db.table(other_table).pks) + if len(other_table_pks) == len(columns): + other_columns = other_table_pks + is_compound = len(rows) > 1 + fks.append( + ForeignKey( + table=self.name, + column=None if is_compound else columns[0], + other_table=other_table, + other_column=None if is_compound else other_columns[0], + columns=columns, + other_columns=other_columns, + is_compound=is_compound, + on_update=rows[0][4], + on_delete=rows[0][5], + ) + ) return fks @property @@ -2141,7 +2329,9 @@ class Table(Queryable): :param pk: New primary key for the table :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 drop_foreign_keys: Foreign key constraints to remove - a column name + drops any foreign key that column participates in, a tuple of column names + drops the compound foreign key with exactly those columns :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 @@ -2226,7 +2416,9 @@ class Table(Queryable): :param pk: New primary key for the table :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 drop_foreign_keys: Foreign key constraints to remove - a column name + drops any foreign key that column participates in, a tuple of column names + drops the compound foreign key with exactly those columns :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 @@ -2253,29 +2445,59 @@ class Table(Queryable): 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, - ) + # Bind fresh names here - type checkers do not narrow captured + # variables inside nested functions, so the closures would + # otherwise see the Optional declared types of drop and rename + dropped_columns = drop + renamed_columns = rename + + def fk_should_be_dropped(fk: ForeignKey) -> bool: + if drop_foreign_keys is not None: + for spec in drop_foreign_keys: + if isinstance(spec, str): + # A column name matches any foreign key it participates in + if spec in fk.columns: + return True + elif tuple(spec) == fk.columns: + # A tuple/list must match a compound key's columns exactly + return True + # Dropping any of a foreign key's columns drops the whole key + return any(column in dropped_columns for column in fk.columns) + + def fk_with_renamed_columns(fk: ForeignKey) -> ForeignKey: + columns = tuple( + renamed_columns.get(column) or column for column in fk.columns + ) + if fk.is_compound: + return ForeignKey( + self.name, + None, + fk.other_table, + None, + columns=columns, + other_columns=fk.other_columns, + is_compound=True, + on_delete=fk.on_delete, + on_update=fk.on_update, ) + return ForeignKey( + self.name, + columns[0], + fk.other_table, + fk.other_columns[0], + on_delete=fk.on_delete, + on_update=fk.on_update, + ) + + create_table_foreign_keys = [] + # Copy over old foreign keys, unless we are dropping them + for fk in self.foreign_keys: + if not fk_should_be_dropped(fk): + create_table_foreign_keys.append(fk_with_renamed_columns(fk)) # 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, - ) - ) + create_table_foreign_keys.append(fk_with_renamed_columns(fk)) new_table_name = "{}_new_{}".format( self.name, tmp_suffix or os.urandom(6).hex() @@ -2708,52 +2930,99 @@ class Table(Queryable): def add_foreign_key( self, - column: str, + column: ForeignKeyColumns, other_table: Optional[str] = None, - other_column: Optional[str] = None, + other_column: Optional[ForeignKeyColumns] = None, ignore: bool = False, + on_delete: str = "NO ACTION", + on_update: str = "NO ACTION", ): """ Alter the schema to mark the specified column as a foreign key to another table. - :param column: The column to mark as a foreign key. + :param column: The column to mark as a foreign key - use a tuple of columns + for a compound foreign key. :param other_table: The table it refers to - if omitted, will be guessed based on the column name. :param other_column: The column on the other table it - if omitted, will be guessed. + Use a tuple of columns for a compound foreign key. :param ignore: Set this to ``True`` to ignore an existing foreign key - otherwise a ``AlterError`` will be raised. + :param on_delete: ``ON DELETE`` action for the foreign key, e.g. ``"CASCADE"`` + or ``"SET NULL"``. + :param on_update: ``ON UPDATE`` action for the foreign key. """ - # Ensure column exists - if column not in self.columns_dict: - raise AlterError("No such column: {}".format(column)) + columns = (column,) if isinstance(column, str) else tuple(column) + # Ensure columns exist + for col in columns: + if col not in self.columns_dict: + raise AlterError("No such column: {}".format(col)) # If other_table is not specified, attempt to guess it from the column if other_table is None: - other_table = self.guess_foreign_table(column) + if len(columns) > 1: + raise ValueError( + "other_table must be specified for a compound foreign key" + ) + other_table = self.guess_foreign_table(columns[0]) # If other_column is not specified, detect the primary key on other_table if other_column is None: - other_column = self.guess_foreign_column(other_table) + if len(columns) > 1: + other_columns = tuple(self.db.table(other_table).pks) + else: + other_columns = (self.guess_foreign_column(other_table),) + elif isinstance(other_column, str): + other_columns = (other_column,) + else: + other_columns = tuple(other_column) + if len(columns) != len(other_columns): + raise ValueError( + "Compound foreign key must have the same number of columns " + "on both sides" + ) - # Soundness check that the other column exists - if ( - not [c for c in self.db[other_table].columns if c.name == other_column] - and other_column != "rowid" - ): - raise AlterError("No such column: {}.{}".format(other_table, other_column)) + # Soundness check that the other columns exist + for other_col in other_columns: + if ( + not [c for c in self.db[other_table].columns if c.name == other_col] + and other_col != "rowid" + ): + raise AlterError("No such column: {}.{}".format(other_table, other_col)) # Check we do not already have an existing foreign key if any( fk for fk in self.foreign_keys - if fk.column == column + if fk.columns == columns and fk.other_table == other_table - and fk.other_column == other_column + and fk.other_columns == other_columns ): if ignore: return self else: raise AlterError( "Foreign key already exists for {} => {}.{}".format( - column, other_table, other_column + ", ".join(columns), other_table, ", ".join(other_columns) ) ) - self.db.add_foreign_keys([(self.name, column, other_table, other_column)]) + if len(columns) == 1: + fk_object = ForeignKey( + self.name, + columns[0], + other_table, + other_columns[0], + on_delete=on_delete, + on_update=on_update, + ) + else: + fk_object = ForeignKey( + self.name, + None, + other_table, + None, + columns=columns, + other_columns=other_columns, + is_compound=True, + on_delete=on_delete, + on_update=on_update, + ) + self.db.add_foreign_keys([fk_object]) return self def enable_counts(self) -> None: diff --git a/tests/test_foreign_keys.py b/tests/test_foreign_keys.py new file mode 100644 index 0000000..8950189 --- /dev/null +++ b/tests/test_foreign_keys.py @@ -0,0 +1,526 @@ +"""Tests for compound (multi-column) foreign keys - issue #594.""" + +import pytest +from sqlite_utils import Database +from sqlite_utils.db import AlterError, ForeignKey +from sqlite_utils.utils import sqlite3 + +COMPOUND_SCHEMA = """ +CREATE TABLE departments ( + campus_name TEXT NOT NULL, + dept_code TEXT NOT NULL, + dept_name TEXT, + PRIMARY KEY (campus_name, dept_code) +); +CREATE TABLE courses ( + course_code TEXT PRIMARY KEY, + course_name TEXT, + campus_name TEXT NOT NULL, + dept_code TEXT NOT NULL, + FOREIGN KEY (campus_name, dept_code) + REFERENCES departments(campus_name, dept_code) +); +""" + + +@pytest.fixture +def compound_db(): + db = Database(memory=True) + db.executescript(COMPOUND_SCHEMA) + return db + + +def test_compound_foreign_key(compound_db): + fks = compound_db["courses"].foreign_keys + assert len(fks) == 1 + fk = fks[0] + assert fk.is_compound is True + assert fk.table == "courses" + assert fk.other_table == "departments" + assert fk.columns == ("campus_name", "dept_code") + assert fk.other_columns == ("campus_name", "dept_code") + # Scalar column/other_column can't sensibly hold a compound key + assert fk.column is None + assert fk.other_column is None + + +def test_single_foreign_key_gets_columns_fields(fresh_db): + fresh_db["authors"].insert({"id": 1, "name": "Sally"}, pk="id") + fresh_db["books"].insert({"title": "Hedgehogs", "author_id": 1}) + fresh_db["books"].add_foreign_key("author_id", "authors", "id") + fk = fresh_db["books"].foreign_keys[0] + assert fk.is_compound is False + assert fk.column == "author_id" + assert fk.other_column == "id" + assert fk.columns == ("author_id",) + assert fk.other_columns == ("id",) + + +def test_foreign_key_no_longer_unpacks_as_tuple(fresh_db): + # Clean break in 4.0: ForeignKey is a dataclass, not a namedtuple, so the + # old tuple unpacking and indexing patterns now fail hard. + fresh_db["authors"].insert({"id": 1, "name": "Sally"}, pk="id") + fresh_db["books"].insert({"title": "Hedgehogs", "author_id": 1}) + fresh_db["books"].add_foreign_key("author_id", "authors", "id") + fk = fresh_db["books"].foreign_keys[0] + with pytest.raises(TypeError): + table, column, other_table, other_column = fk + with pytest.raises(TypeError): + fk[0] + + +def test_foreign_keys_are_sortable(fresh_db): + fresh_db["authors"].insert({"id": 1, "name": "Sally"}, pk="id") + fresh_db["categories"].insert({"id": 1, "name": "Wildlife"}, pk="id") + fresh_db["books"].insert({"title": "Hedgehogs", "author_id": 1, "category_id": 1}) + fresh_db.add_foreign_keys( + [ + ("books", "author_id", "authors", "id"), + ("books", "category_id", "categories", "id"), + ] + ) + fks = sorted(fresh_db["books"].foreign_keys) + assert fks[0].column == "author_id" + assert fks[1].column == "category_id" + + +def test_mixed_compound_and_single_foreign_keys_are_sortable(): + # compound FKs have column=None, which must not break sorting + # against single-column FKs (None < str raises TypeError) + db = Database(memory=True) + db.executescript(""" + CREATE TABLE departments ( + campus_name TEXT NOT NULL, + dept_code TEXT NOT NULL, + PRIMARY KEY (campus_name, dept_code) + ); + CREATE TABLE accreditations (id INTEGER PRIMARY KEY); + CREATE TABLE courses ( + course_code TEXT PRIMARY KEY, + campus_name TEXT NOT NULL, + dept_code TEXT NOT NULL, + accreditation_id INTEGER REFERENCES accreditations(id), + FOREIGN KEY (campus_name, dept_code) + REFERENCES departments(campus_name, dept_code) + ); + """) + fks = db["courses"].foreign_keys + assert len(fks) == 2 + assert {fk.is_compound for fk in fks} == {True, False} + fks_sorted = sorted(fks) + assert fks_sorted[0].other_table == "accreditations" + assert fks_sorted[1].other_table == "departments" + + +@pytest.fixture +def departments_db(): + db = Database(memory=True) + db.create_table( + "departments", + {"campus_name": str, "dept_code": str, "dept_name": str}, + pk=("campus_name", "dept_code"), + ) + return db + + +EXPECTED_COURSES_SCHEMA = ( + 'CREATE TABLE "courses" (\n' + ' "course_code" TEXT PRIMARY KEY,\n' + ' "campus_name" TEXT,\n' + ' "dept_code" TEXT,\n' + ' FOREIGN KEY ("campus_name", "dept_code") ' + 'REFERENCES "departments"("campus_name", "dept_code")\n' + ")" +) + + +@pytest.mark.parametrize( + "foreign_keys", + ( + [ + ForeignKey( + table="courses", + column=None, + other_table="departments", + other_column=None, + columns=("campus_name", "dept_code"), + other_columns=("campus_name", "dept_code"), + is_compound=True, + ) + ], + [(("campus_name", "dept_code"), "departments", ("campus_name", "dept_code"))], + # Two-item form guesses the other table's primary key: + [(("campus_name", "dept_code"), "departments")], + # Lists work too, though tuples are the documented form: + [(["campus_name", "dept_code"], "departments", ["campus_name", "dept_code"])], + ), +) +def test_create_table_with_compound_foreign_key(departments_db, foreign_keys): + departments_db.create_table( + "courses", + {"course_code": str, "campus_name": str, "dept_code": str}, + pk="course_code", + foreign_keys=foreign_keys, + ) + assert departments_db["courses"].schema == EXPECTED_COURSES_SCHEMA + fks = departments_db["courses"].foreign_keys + assert len(fks) == 1 + fk = fks[0] + assert fk.is_compound is True + assert fk.columns == ("campus_name", "dept_code") + assert fk.other_table == "departments" + assert fk.other_columns == ("campus_name", "dept_code") + + +def test_create_table_compound_foreign_key_enforced(departments_db): + departments_db.execute("PRAGMA foreign_keys = ON") + departments_db.create_table( + "courses", + {"course_code": str, "campus_name": str, "dept_code": str}, + pk="course_code", + foreign_keys=[(("campus_name", "dept_code"), "departments")], + ) + departments_db["departments"].insert( + {"campus_name": "Berkeley", "dept_code": "CS", "dept_name": "Computer Science"} + ) + departments_db["courses"].insert( + {"course_code": "CS101", "campus_name": "Berkeley", "dept_code": "CS"} + ) + with pytest.raises(sqlite3.IntegrityError): + departments_db.execute( + "insert into courses (course_code, campus_name, dept_code) " + "values ('X1', 'Nowhere', 'NOPE')" + ) + + +def test_create_table_compound_foreign_key_missing_other_column(departments_db): + with pytest.raises(AlterError): + departments_db.create_table( + "courses", + {"course_code": str, "campus_name": str, "dept_code": str}, + pk="course_code", + foreign_keys=[ + (("campus_name", "dept_code"), "departments", ("campus_name", "nope")) + ], + ) + + +def test_transform_preserves_compound_foreign_key(compound_db): + compound_db["courses"].transform(rename={"course_name": "title"}) + fks = compound_db["courses"].foreign_keys + assert len(fks) == 1 + fk = fks[0] + assert fk.is_compound is True + assert fk.columns == ("campus_name", "dept_code") + assert fk.other_table == "departments" + assert fk.other_columns == ("campus_name", "dept_code") + + +def test_transform_rename_member_column_updates_compound_foreign_key(compound_db): + compound_db["courses"].transform(rename={"campus_name": "campus"}) + fks = compound_db["courses"].foreign_keys + assert len(fks) == 1 + fk = fks[0] + assert fk.is_compound is True + assert fk.columns == ("campus", "dept_code") + # Referenced columns in the other table are unchanged + assert fk.other_columns == ("campus_name", "dept_code") + + +def test_transform_drop_member_column_drops_compound_foreign_key(compound_db): + # Matches single-column behavior: dropping the column silently + # drops the foreign key that used it + compound_db["courses"].transform(drop={"dept_code"}) + assert compound_db["courses"].foreign_keys == [] + assert "FOREIGN KEY" not in compound_db["courses"].schema + + +@pytest.mark.parametrize( + "drop_foreign_keys", + ( + # A bare column name matches any foreign key it participates in: + ["campus_name"], + # A tuple must match the full compound key: + [("campus_name", "dept_code")], + ), +) +def test_transform_drop_compound_foreign_key(compound_db, drop_foreign_keys): + compound_db["courses"].transform(drop_foreign_keys=drop_foreign_keys) + assert compound_db["courses"].foreign_keys == [] + # The columns themselves survive + assert {"campus_name", "dept_code"} <= set( + compound_db["courses"].columns_dict.keys() + ) + + +@pytest.fixture +def courses_db(departments_db): + departments_db.create_table( + "courses", + {"course_code": str, "campus_name": str, "dept_code": str}, + pk="course_code", + ) + return departments_db + + +def test_add_compound_foreign_key(courses_db): + t = courses_db["courses"].add_foreign_key( + ("campus_name", "dept_code"), "departments", ("campus_name", "dept_code") + ) + # Returns self + assert t.name == "courses" + fks = courses_db["courses"].foreign_keys + assert len(fks) == 1 + fk = fks[0] + assert fk.is_compound is True + assert fk.columns == ("campus_name", "dept_code") + assert fk.other_table == "departments" + assert fk.other_columns == ("campus_name", "dept_code") + + +def test_add_compound_foreign_key_guesses_other_columns(courses_db): + # Lists work here too, though tuples are the documented form + courses_db["courses"].add_foreign_key(["campus_name", "dept_code"], "departments") + fk = courses_db["courses"].foreign_keys[0] + assert fk.other_columns == ("campus_name", "dept_code") + + +def test_add_compound_foreign_key_error_if_already_exists(courses_db): + courses_db["courses"].add_foreign_key(("campus_name", "dept_code"), "departments") + with pytest.raises(AlterError) as ex: + courses_db["courses"].add_foreign_key( + ("campus_name", "dept_code"), "departments" + ) + assert "already exists" in ex.value.args[0] + # ignore=True should not raise + courses_db["courses"].add_foreign_key( + ("campus_name", "dept_code"), "departments", ignore=True + ) + + +def test_add_compound_foreign_key_error_if_column_missing(courses_db): + with pytest.raises(AlterError): + courses_db["courses"].add_foreign_key(("campus_name", "nope"), "departments") + + +def test_db_add_foreign_keys_compound(courses_db): + courses_db.add_foreign_keys( + [ + ( + "courses", + ("campus_name", "dept_code"), + "departments", + ("campus_name", "dept_code"), + ) + ] + ) + fk = courses_db["courses"].foreign_keys[0] + assert fk.is_compound is True + assert fk.columns == ("campus_name", "dept_code") + + +def test_index_foreign_keys_compound_creates_composite_index(compound_db): + compound_db.index_foreign_keys() + index_columns = [i.columns for i in compound_db["courses"].indexes] + assert ["campus_name", "dept_code"] in index_columns + # No separate single-column indexes for the members + assert ["campus_name"] not in index_columns + assert ["dept_code"] not in index_columns + + +def test_foreign_key_captures_on_delete_and_on_update(): + db = Database(memory=True) + db.executescript(""" + CREATE TABLE authors (id INTEGER PRIMARY KEY); + CREATE TABLE books ( + id INTEGER PRIMARY KEY, + author_id INTEGER REFERENCES authors(id) + ON DELETE CASCADE ON UPDATE RESTRICT + ); + """) + fk = db["books"].foreign_keys[0] + assert fk.on_delete == "CASCADE" + assert fk.on_update == "RESTRICT" + + +def test_foreign_key_on_delete_defaults_to_no_action(fresh_db): + fresh_db["authors"].insert({"id": 1}, pk="id") + fresh_db["books"].insert({"id": 1, "author_id": 1}, pk="id") + fresh_db["books"].add_foreign_key("author_id", "authors", "id") + fk = fresh_db["books"].foreign_keys[0] + assert fk.on_delete == "NO ACTION" + assert fk.on_update == "NO ACTION" + + +def test_create_table_foreign_key_with_on_delete(fresh_db): + fresh_db["authors"].insert({"id": 1}, pk="id") + fresh_db.create_table( + "books", + {"id": int, "author_id": int}, + pk="id", + foreign_keys=[ + ForeignKey( + table="books", + column="author_id", + other_table="authors", + other_column="id", + on_delete="CASCADE", + ) + ], + ) + assert "ON DELETE CASCADE" in fresh_db["books"].schema + assert fresh_db["books"].foreign_keys[0].on_delete == "CASCADE" + + +def test_transform_preserves_on_delete_cascade(): + db = Database(memory=True) + db.executescript(""" + CREATE TABLE authors (id INTEGER PRIMARY KEY); + CREATE TABLE books ( + id INTEGER PRIMARY KEY, + title TEXT, + author_id INTEGER REFERENCES authors(id) ON DELETE CASCADE + ); + """) + db["books"].transform(rename={"title": "book_title"}) + fk = db["books"].foreign_keys[0] + assert fk.on_delete == "CASCADE" + assert fk.on_update == "NO ACTION" + assert "ON DELETE CASCADE" in db["books"].schema + + +def test_transform_preserves_compound_foreign_key_on_delete(): + db = Database(memory=True) + db.executescript(""" + CREATE TABLE departments ( + campus_name TEXT NOT NULL, + dept_code TEXT NOT NULL, + PRIMARY KEY (campus_name, dept_code) + ); + CREATE TABLE courses ( + course_code TEXT PRIMARY KEY, + campus_name TEXT NOT NULL, + dept_code TEXT NOT NULL, + FOREIGN KEY (campus_name, dept_code) + REFERENCES departments(campus_name, dept_code) ON DELETE CASCADE + ); + """) + db["courses"].transform(rename={"course_code": "code"}) + fk = db["courses"].foreign_keys[0] + assert fk.is_compound is True + assert fk.on_delete == "CASCADE" + assert "ON DELETE CASCADE" in db["courses"].schema + + +def test_implicit_primary_key_reference_is_resolved(): + # REFERENCES authors (no column) has "to" of None in the pragma - + # it should be resolved to the primary key of the other table + db = Database(memory=True) + db.executescript(""" + CREATE TABLE authors (author_id INTEGER PRIMARY KEY); + CREATE TABLE books ( + id INTEGER PRIMARY KEY, + author_id INTEGER REFERENCES authors + ); + """) + fk = db["books"].foreign_keys[0] + assert fk.is_compound is False + assert fk.other_column == "author_id" + assert fk.other_columns == ("author_id",) + + +def test_implicit_compound_primary_key_reference_is_resolved(): + db = Database(memory=True) + db.executescript(""" + CREATE TABLE departments ( + campus_name TEXT NOT NULL, + dept_code TEXT NOT NULL, + PRIMARY KEY (campus_name, dept_code) + ); + CREATE TABLE courses ( + course_code TEXT PRIMARY KEY, + campus_name TEXT NOT NULL, + dept_code TEXT NOT NULL, + FOREIGN KEY (campus_name, dept_code) REFERENCES departments + ); + """) + fk = db["courses"].foreign_keys[0] + assert fk.is_compound is True + assert fk.other_columns == ("campus_name", "dept_code") + + +def test_foreign_key_normalizes_list_columns_to_tuples(): + # Compound columns passed as lists are normalized to tuples, so they + # compare equal to introspected ForeignKeys + fk = ForeignKey( + table="courses", + column=None, + other_table="departments", + other_column=None, + columns=["campus_name", "dept_code"], + other_columns=["campus_name", "dept_code"], + is_compound=True, + ) + assert fk.columns == ("campus_name", "dept_code") + assert fk.other_columns == ("campus_name", "dept_code") + + +def test_add_foreign_keys_preserves_actions(fresh_db): + # https://github.com/simonw/sqlite-utils/issues/594 review finding: + # ForeignKey objects passed to db.add_foreign_keys() were flattened + # to plain tuples, losing on_delete/on_update + fresh_db["authors"].insert({"id": 1}, pk="id") + fresh_db["books"].insert({"id": 1, "author_id": 1}, pk="id") + fresh_db.add_foreign_keys( + [ForeignKey("books", "author_id", "authors", "id", on_delete="CASCADE")] + ) + fk = fresh_db["books"].foreign_keys[0] + assert fk.on_delete == "CASCADE" + assert "ON DELETE CASCADE" in fresh_db["books"].schema + + +def test_add_foreign_keys_preserves_actions_compound(courses_db): + courses_db.add_foreign_keys( + [ + ForeignKey( + table="courses", + column=None, + other_table="departments", + other_column=None, + columns=("campus_name", "dept_code"), + other_columns=("campus_name", "dept_code"), + is_compound=True, + on_delete="CASCADE", + ) + ] + ) + fk = courses_db["courses"].foreign_keys[0] + assert fk.is_compound is True + assert fk.on_delete == "CASCADE" + assert "ON DELETE CASCADE" in courses_db["courses"].schema + + +def test_add_foreign_key_on_delete_on_update(fresh_db): + fresh_db["authors"].insert({"id": 1}, pk="id") + fresh_db["books"].insert({"id": 1, "author_id": 1}, pk="id") + fresh_db["books"].add_foreign_key( + "author_id", "authors", "id", on_delete="CASCADE", on_update="RESTRICT" + ) + fk = fresh_db["books"].foreign_keys[0] + assert fk.on_delete == "CASCADE" + assert fk.on_update == "RESTRICT" + assert "ON UPDATE RESTRICT ON DELETE CASCADE" in fresh_db["books"].schema + # The cascade should actually fire + fresh_db.execute("PRAGMA foreign_keys = ON") + fresh_db.execute("delete from authors where id = 1") + assert fresh_db["books"].count == 0 + + +def test_add_compound_foreign_key_on_delete(courses_db): + courses_db["courses"].add_foreign_key( + ("campus_name", "dept_code"), "departments", on_delete="SET NULL" + ) + fk = courses_db["courses"].foreign_keys[0] + assert fk.is_compound is True + assert fk.on_delete == "SET NULL" + assert "ON DELETE SET NULL" in courses_db["courses"].schema