remote/datasette
1
0
Fork 0
mirror of https://github.com/simonw/datasette.git synced 2026-10-25 17:56:57 +01:00
Code Issues Projects Releases Packages Wiki Activity

test: keep WITHOUT ROWID readback within SQLite variable limits

Browse source 
Query inserted rows by each explicit primary key tuple to preserve request order and avoid a combined query exceeding older SQLite parameter limits.
This commit is contained in:
kokokoXUY 2026-09-26 12:04:37 +08:00
commit ad20db2ef8
2 changed files with 32 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

17
datasette/views/table.py
View file

@ -1147,16 +1147,15 @@ class TableInsertView(BaseView):
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
where_clause = " AND ".join(
f"{escape_sqlite(pk)} = ?" for pk in pks
)
args = [row[pk] for row in rows for pk in pks]
return list(table.rows_where(where_clause, args))
returned_rows = []
for row in rows:
returned_rows.extend(
table.rows_where(where_clause, [row[pk] for pk in pks])
)
return returned_rows
return list(
table.rows_where(
"rowid in ({})".format(",".join("?" for _ in rowids)),

24
tests/test_api_write.py
View file

@ -487,6 +487,30 @@ async def test_insert_rows_into_without_rowid_table_preserves_return_order(ds_wr
assert response.json()["rows"] == rows
@pytest.mark.asyncio
@pytest.mark.skipif(
not hasattr(sqlite3.Connection, "setlimit"),
reason="SQLite variable limit controls require Python 3.11",
)
async def test_insert_rows_into_without_rowid_table_respects_variable_limit(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"
)
await db.execute_write_fn(
lambda conn: conn.setlimit(sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER, 4)
)
rows = [{"a": str(i), "b": "value"} for i in range(3)]
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)