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

Compound foreign key support (PR #770)

Browse source 
Closes #594, closes #530

- **Breaking:** `ForeignKey` is now a dataclass, not a namedtuple - tuple unpacking/indexing raises `TypeError`; access fields by name. New fields: `columns`, `other_columns`, `is_compound`, `on_delete`, `on_update`.
- `table.foreign_keys` returns a compound foreign key as a single `ForeignKey` with `is_compound=True`, instead of one misleading entry per column.
- Create compound foreign keys anywhere with tuples of columns: `foreign_keys=[(("campus_name", "dept_code"), "departments")]` - also in `add_foreign_key()` and `db.add_foreign_keys()`. Rendered as table-level `FOREIGN KEY` constraints.
- `transform()` round-trips compound keys: `rename=` applies to member columns, dropping a member column drops the constraint, `drop_foreign_keys=` accepts a column name or an exact column tuple.
- `ON DELETE`/`ON UPDATE` actions are captured, rendered and preserved - `transform()` no longer strips `ON DELETE CASCADE`. New `add_foreign_key(..., on_delete="CASCADE")` parameters.
- Fixes: sorting mixed compound/single keys no longer raises `TypeError`; implicit `REFERENCES other_table` resolves to the other table's primary key; `index_foreign_keys()` creates a composite index for compound keys.
This commit is contained in:
Simon Willison 2026-07-05 15:54:44 -07:00 • committed by GitHub
commit 9a3936531d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 1073 additions and 120 deletions
Download patch file Download diff file Expand all files Collapse all files

3
Justfile
View file

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

23
docs/changelog.rst
View file

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

123
docs/python-api.rst
View file

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

7
docs/reference.rst
View file

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

18
docs/upgrading.rst
View file

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

491
sqlite_utils/db.py
View file

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

526
tests/test_foreign_keys.py Normal file
View file

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