Canned query writes support JSON POST body, refs #880

datasette/utils/testing.py

@@ -50,6 +50,7 @@ class TestClient:
         self,
         path,
         post_data=None,
+        body=None,
         allow_redirects=True,
         redirect_count=0,
         content_type="application/x-www-form-urlencoded",
@@ -58,21 +59,25 @@ class TestClient:
     ):
         cookies = cookies or {}
         post_data = post_data or {}
+        assert not (post_data and body), "Provide one or other of body= or post_data="
         # Maybe fetch a csrftoken first
         if csrftoken_from is not None:
+            assert body is None, "body= is not compatible with csrftoken_from="
             if csrftoken_from is True:
                 csrftoken_from = path
             token_response = await self._request(csrftoken_from, cookies=cookies)
             csrftoken = token_response.cookies["ds_csrftoken"]
             cookies["ds_csrftoken"] = csrftoken
             post_data["csrftoken"] = csrftoken
+        if post_data:
+            body = urlencode(post_data, doseq=True)
         return await self._request(
             path,
             allow_redirects,
             redirect_count,
             "POST",
             cookies,
-            post_data,
+            body,
             content_type,
         )
 
@@ -83,7 +88,7 @@ class TestClient:
         redirect_count=0,
         method="GET",
         cookies=None,
-        post_data=None,
+        post_body=None,
         content_type=None,
     ):
         query_string = b""
@@ -113,8 +118,8 @@ class TestClient:
         }
         instance = ApplicationCommunicator(self.asgi_app, scope)
 
-        if post_data:
-            body = urlencode(post_data, doseq=True).encode("utf-8")
+        if post_body:
+            body = post_body.encode("utf-8")
             await instance.send_input({"type": "http.request", "body": body})
         else:
             await instance.send_input({"type": "http.request"})

datasette/views/database.py

@@ -1,6 +1,8 @@
 import os
 import itertools
 import jinja2
+import json
+from urllib.parse import parse_qsl
 
 from datasette.utils import (
     check_visibility,
@@ -208,7 +210,15 @@ class QueryView(DataView):
         # Execute query - as write or as read
         if write:
             if request.method == "POST":
-                params = await request.post_vars()
+                body = await request.post_body()
+                body = body.decode("utf-8").strip()
+                if body.startswith("{") and body.endswith("}"):
+                    params = json.loads(body)
+                    # But we want key=value strings
+                    for key, value in params.items():
+                        params[key] = str(value)
+                else:
+                    params = dict(parse_qsl(body, keep_blank_values=True))
                 if canned_query:
                     params_for_query = MagicParameters(params, request, self.ds)
                 else:

tests/test_canned_queries.py

@@ -1,4 +1,5 @@
 from bs4 import BeautifulSoup as Soup
+import json
 import pytest
 import re
 from .fixtures import make_app_client, app_client
@@ -163,6 +164,16 @@ def test_vary_header(canned_write_client):
     assert "Cookie" == canned_write_client.get("/data/update_name").headers["vary"]
 
 
+def test_json_post_body(canned_write_client):
+    response = canned_write_client.post(
+        "/data/add_name",
+        body=json.dumps({"name": "Hello"}),
+        allow_redirects=False,
+    )
+    assert 302 == response.status
+    assert "/data/add_name?success" == response.headers["Location"]
+
+
 def test_canned_query_permissions_on_database_page(canned_write_client):
     # Without auth only shows three queries
     query_names = {