diff --git a/docs/python-api.rst b/docs/python-api.rst index 882f952..f44a10b 100644 --- a/docs/python-api.rst +++ b/docs/python-api.rst @@ -853,6 +853,25 @@ As with single columns, you can leave off the list of other columns to reference (["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 @@ -2280,6 +2299,10 @@ Each ``ForeignKey`` has the following attributes: A list 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. diff --git a/sqlite_utils/db.py b/sqlite_utils/db.py index 3c78e3b..1910b84 100644 --- a/sqlite_utils/db.py +++ b/sqlite_utils/db.py @@ -190,6 +190,8 @@ class ForeignKey: columns: List[str] = field(default_factory=list) other_columns: List[str] = field(default_factory=list) 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 @@ -201,6 +203,16 @@ class ForeignKey: ) +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( @@ -1345,14 +1357,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( - cast(str, 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] @@ -1385,12 +1395,13 @@ class Database: "No such column: {}".format(", ".join(sorted(missing))) ) column_defs.append( - " FOREIGN KEY ({columns}) REFERENCES {other_table}({other_columns})".format( + " 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) @@ -2055,7 +2066,9 @@ class Table(Queryable): ).fetchall(): if row is not None: id, seq, table_name, from_, to_, on_update, on_delete, match = row - by_id.setdefault(id, []).append((seq, table_name, from_, 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 @@ -2072,6 +2085,8 @@ class Table(Queryable): columns=columns, other_columns=other_columns, is_compound=is_compound, + on_update=rows[0][4], + on_delete=rows[0][5], ) ) return fks @@ -2434,9 +2449,16 @@ class Table(Queryable): 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] + 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 = [] diff --git a/tests/test_foreign_keys.py b/tests/test_foreign_keys.py index 1fabce7..afe333f 100644 --- a/tests/test_foreign_keys.py +++ b/tests/test_foreign_keys.py @@ -324,3 +324,87 @@ def test_index_foreign_keys_compound_creates_composite_index(compound_db): # 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