fix: return inserted rows from WITHOUT ROWID tables

Look up returned rows by explicit primary keys when SQLite tables have no rowid. Cover compound keys and STRICT table options. Refs #1873.

datasette/views/table.py

@@ -1,6 +1,7 @@
 import asyncio
 import itertools
 import json
+import re
 import time
 import urllib
 import urllib.parse
@@ -1144,6 +1145,18 @@ class TableInsertView(BaseView):
                 method = table.upsert if upsert else table.insert
                 for row in rows:
                     rowids.append(method(row, **kwargs).last_rowid)
+                table_options = table.schema.rsplit(")", 1)[-1]
+                if re.search(r"\bWITHOUT\s+ROWID\b", table_options, re.IGNORECASE):
+                    if not rows:
+                        return []
+                    where_clause = " OR ".join(
+                        "({})".format(
+                            " AND ".join(f"{escape_sqlite(pk)} = ?" for pk in pks)
+                        )
+                        for _ in rows
+                    )
+                    args = [row[pk] for row in rows for pk in pks]
+                    return list(table.rows_where(where_clause, args))
                 return list(
                     table.rows_where(
                         "rowid in ({})".format(",".join("?" for _ in rowids)),

docs/json_api.rst

@@ -1827,6 +1827,9 @@ If successful, this will return a ``201`` status code and the newly inserted row
         ]
     }
 
+This also works for SQLite ``WITHOUT ROWID`` tables with explicit primary key
+values. The returned ``"rows"`` list is looked up using those primary keys.
+
 To insert multiple rows at a time, use the same API method but send a list of dictionaries as the ``"rows"`` key:
 
 ::

tests/test_api_write.py

@@ -438,6 +438,55 @@ async def test_insert_row(ds_write, content_type):
     assert not event.replace
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "table_options",
+    (
+        "without rowid",
+        pytest.param(
+            "without rowid, strict",
+            marks=pytest.mark.skipif(
+                sqlite3.sqlite_version_info < (3, 37, 0),
+                reason="STRICT tables require SQLite 3.37",
+            ),
+        ),
+    ),
+)
+async def test_insert_row_into_without_rowid_table(ds_write, table_options):
+    db = ds_write.get_database("data")
+    await db.execute_write(
+        "create table compound_without_rowid (a text, b text, "
+        f"primary key (a, b)) {table_options}"
+    )
+    response = await ds_write.client.post(
+        "/data/compound_without_rowid/-/insert",
+        json={"row": {"a": "first", "b": "second"}},
+        headers=_headers(write_token(ds_write)),
+    )
+    assert response.status_code == 201, response.json()
+    assert response.json()["rows"] == [{"a": "first", "b": "second"}]
+    assert (await db.execute("select a, b from compound_without_rowid")).dicts() == [
+        {"a": "first", "b": "second"}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_insert_rows_into_without_rowid_table_preserves_return_order(ds_write):
+    db = ds_write.get_database("data")
+    await db.execute_write(
+        "create table compound_without_rowid (a text, b text, "
+        "primary key (a, b)) without rowid"
+    )
+    rows = [{"a": "z", "b": "first"}, {"a": "a", "b": "second"}]
+    response = await ds_write.client.post(
+        "/data/compound_without_rowid/-/insert",
+        json={"rows": rows, "return": True},
+        headers=_headers(write_token(ds_write)),
+    )
+    assert response.status_code == 201, response.json()
+    assert response.json()["rows"] == rows
+
+
 @pytest.mark.asyncio
 async def test_insert_row_alter(ds_write):
     token = write_token(ds_write)