Merge c9a365e1ef into c435a80030

datasette/views/table_create_alter.py

@@ -935,7 +935,27 @@ 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.
+                # Matched case-insensitively because SQLite resolves identifiers
+                # that way: pk "code" against a column named "Code" still creates
+                # that column as the primary key.
+                pk_names = {
+                    name.lower() for name in (pks if pks else ([pk] if pk else []))
+                }
+                if len(pk_names) == 1 and any(
+                    column.name.lower() 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.lower() in pk_names
+                ]
                 defaults = {}
                 for column in columns:
                     if "default_expr" in column.model_fields_set:

tests/test_api_write.py

@@ -2489,6 +2489,113 @@ 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
+
+    # Primary key named with different casing from the column. SQLite resolves
+    # identifiers case-insensitively, so this still creates "Code" as the primary
+    # key and it must pick up NOT NULL too.
+    response = await ds_write.client.post(
+        "/data/-/create",
+        json={
+            "table": "pk_mixed_case",
+            "columns": [
+                {"name": "Code", "type": "text"},
+                {"name": "name", "type": "text"},
+            ],
+            "pk": "code",
+        },
+        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_mixed_case')")
+        ).dicts()
+    }
+    assert columns["Code"]["pk"] == 1
+    assert columns["Code"]["notnull"] == 1
+    assert columns["name"]["notnull"] == 0
+
+    # The integer-rowid-alias exemption is matched case-insensitively too, so an
+    # integer pk stays a plain rowid alias rather than gaining a constraint.
+    response = await ds_write.client.post(
+        "/data/-/create",
+        json={
+            "table": "pk_mixed_case_integer",
+            "columns": [
+                {"name": "Id", "type": "integer"},
+                {"name": "name", "type": "text"},
+            ],
+            "pk": "id",
+        },
+        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_mixed_case_integer')")
+        ).dicts()
+    }
+    assert columns["Id"]["pk"] == 1
+    assert columns["Id"]["notnull"] == 0
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "default_expr,minimum_value,expected_schema",