From 3956e6365006f6ff0cc3a67a0b9c8f0c5ea9f055 Mon Sep 17 00:00:00 2001
From: Pranav Mishra <pranav@harperinsure.com>
Date: Tue, 7 Jul 2026 13:46:16 -0700
Subject: [PATCH] Ensure primary keys created via the JSON API are NOT NULL

The table-create JSON API built its own CREATE TABLE without marking
primary key columns NOT NULL, so a text or composite primary key could
be created nullable. SQLite then allows rows with a NULL primary key,
which cannot be viewed, edited or deleted. Force primary key columns
NOT NULL, leaving a lone integer primary key alone since it aliases the
rowid and is never NULL.

Refs #2807

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 datasette/views/table_create_alter.py | 17 +++++++-
 tests/test_api_write.py               | 57 +++++++++++++++++++++++++++
 2 files changed, 73 insertions(+), 1 deletion(-)

diff --git a/datasette/views/table_create_alter.py b/datasette/views/table_create_alter.py
index 4deeafcc..61ceabd2 100644
--- a/datasette/views/table_create_alter.py
+++ b/datasette/views/table_create_alter.py
@@ -900,7 +900,22 @@ class TableCreateView(BaseView):
                     rows, pk=pks or pk, ignore=ignore, replace=replace, alter=alter
                 )
             else:
-                not_null = [column.name for column in columns if column.not_null]
+                # Force primary key columns NOT NULL so the API cannot create
+                # NULL primary keys, which SQLite otherwise allows and which
+                # leaves rows that cannot be viewed, edited or deleted (see
+                # #2807). A single integer primary key aliases the rowid and is
+                # never NULL, so it needs no explicit constraint.
+                pk_names = set(pks) if pks else ({pk} if pk else set())
+                if len(pk_names) == 1 and any(
+                    column.name in pk_names and column.type == "integer"
+                    for column in columns
+                ):
+                    pk_names = set()
+                not_null = [
+                    column.name
+                    for column in columns
+                    if column.not_null or column.name in pk_names
+                ]
                 defaults = {}
                 for column in columns:
                     if "default_expr" in column.model_fields_set:
diff --git a/tests/test_api_write.py b/tests/test_api_write.py
index 840bd05b..f531e41a 100644
--- a/tests/test_api_write.py
+++ b/tests/test_api_write.py
@@ -2296,6 +2296,63 @@ async def test_create_table_with_column_constraints(ds_write):
     assert columns[4]["dflt_value"] == "'hello)'"
 
 
+@pytest.mark.asyncio
+async def test_create_table_primary_keys_are_not_null(ds_write):
+    # Primary keys created via the JSON API must be NOT NULL - see #2807
+    token = write_token(ds_write)
+    db = ds_write.get_database("data")
+
+    # Single (non-integer) primary key
+    response = await ds_write.client.post(
+        "/data/-/create",
+        json={
+            "table": "pk_single",
+            "columns": [
+                {"name": "code", "type": "text"},
+                {"name": "name", "type": "text"},
+            ],
+            "pk": "code",
+        },
+        headers=_headers(token),
+    )
+    assert response.status_code == 201, response.text
+    assert "PRIMARY KEY NOT NULL" in response.json()["schema"]
+    columns = {
+        column["name"]: column
+        for column in (
+            await db.execute("select * from pragma_table_info('pk_single')")
+        ).dicts()
+    }
+    assert columns["code"]["pk"] == 1
+    assert columns["code"]["notnull"] == 1
+    assert columns["name"]["notnull"] == 0
+
+    # Composite primary key - every key column is NOT NULL
+    response = await ds_write.client.post(
+        "/data/-/create",
+        json={
+            "table": "pk_composite",
+            "columns": [
+                {"name": "a", "type": "text"},
+                {"name": "b", "type": "text"},
+                {"name": "value", "type": "text"},
+            ],
+            "pks": ["a", "b"],
+        },
+        headers=_headers(token),
+    )
+    assert response.status_code == 201, response.text
+    columns = {
+        column["name"]: column
+        for column in (
+            await db.execute("select * from pragma_table_info('pk_composite')")
+        ).dicts()
+    }
+    assert columns["a"]["notnull"] == 1
+    assert columns["b"]["notnull"] == 1
+    assert columns["value"]["notnull"] == 0
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "default_expr,minimum_value,expected_schema",