Capture and preserve foreign key ON DELETE/ON UPDATE actions

ForeignKey gains on_delete and on_update fields (default "NO ACTION"),
populated from PRAGMA foreign_key_list. create_table_sql() renders the
corresponding clauses for both inline and compound table-level foreign
keys, which means table.transform() now preserves actions such as
ON DELETE CASCADE - previously they were silently stripped whenever a
table was transformed.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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

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 = []

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