diff --git a/datasette/templates/query.html b/datasette/templates/query.html
index 2c8c05a0..52896e96 100644
--- a/datasette/templates/query.html
+++ b/datasette/templates/query.html
@@ -27,11 +27,12 @@
 {% endblock %}
 
 {% block content %}
+
 <h1 style="padding-left: 10px; border-left: 10px solid #{{ database_color(database) }}">{{ metadata.title or database }}</h1>
 
 {% block description_source_license %}{% include "_description_source_license.html" %}{% endblock %}
 
-<form class="sql" action="{{ database_url(database) }}{% if canned_query %}/{{ canned_query }}{% endif %}" method="get">
+<form class="sql" action="{{ database_url(database) }}{% if canned_query %}/{{ canned_query }}{% endif %}" method="{% if canned_write %}post{% else %}get{% endif %}">
     <h3>Custom SQL query{% if display_rows %} returning {% if truncated %}more than {% endif %}{{ "{:,}".format(display_rows|length) }} row{% if display_rows|length == 1 %}{% else %}s{% endif %}{% endif %} <span class="show-hide-sql">{% if hide_sql %}(<a href="{{ path_with_removed_args(request, {'_hide_sql': '1'}) }}">show</a>){% else %}(<a href="{{ path_with_added_args(request, {'_hide_sql': '1'}) }}">hide</a>){% endif %}</span></h3>
     {% if not hide_sql %}
         {% if editable and config.allow_sql %}
@@ -74,7 +75,9 @@
     </tbody>
 </table>
 {% else %}
-    <p class="zero-results">0 results</p>
+    {% if not canned_write %}
+        <p class="zero-results">0 results</p>
+    {% endif %}
 {% endif %}
 
 {% include "_codemirror_foot.html" %}
diff --git a/datasette/views/database.py b/datasette/views/database.py
index 15545fb8..558dd0f0 100644
--- a/datasette/views/database.py
+++ b/datasette/views/database.py
@@ -106,6 +106,8 @@ class QueryView(DataView):
         canned_query=None,
         metadata=None,
         _size=None,
+        named_parameters=None,
+        write=False,
     ):
         params = {key: request.args.get(key) for key in request.args}
         if "sql" in params:
@@ -113,7 +115,7 @@ class QueryView(DataView):
         if "_shape" in params:
             params.pop("_shape")
         # Extract any :named parameters
-        named_parameters = self.re_named_parameter.findall(sql)
+        named_parameters = named_parameters or self.re_named_parameter.findall(sql)
         named_parameter_values = {
             named_parameter: params.get(named_parameter) or ""
             for named_parameter in named_parameters
@@ -129,12 +131,60 @@ class QueryView(DataView):
             extra_args["custom_time_limit"] = int(params["_timelimit"])
         if _size:
             extra_args["page_size"] = _size
-        results = await self.ds.execute(
-            database, sql, params, truncate=True, **extra_args
-        )
-        columns = [r[0] for r in results.description]
 
         templates = ["query-{}.html".format(to_css_class(database)), "query.html"]
+
+        # Execute query - as write or as read
+        if write:
+            if request.method == "POST":
+                params = await request.post_vars()
+                try:
+                    cursor = await self.ds.databases[database].execute_write(
+                        sql, params, block=True
+                    )
+                    message = metadata.get(
+                        "on_success_message"
+                    ) or "Query executed, {} row{} affected".format(
+                        cursor.rowcount, "" if cursor.rowcount == 1 else "s"
+                    )
+                    message_type = self.ds.INFO
+                    redirect_url = metadata.get("on_success_redirect")
+                except Exception as e:
+                    message = metadata.get("on_error_message") or str(e)
+                    message_type = self.ds.ERROR
+                    redirect_url = metadata.get("on_error_redirect")
+                self.ds.add_message(request, message, message_type)
+                return self.redirect(request, redirect_url or request.path)
+            else:
+
+                async def extra_template():
+                    return {
+                        "request": request,
+                        "path_with_added_args": path_with_added_args,
+                        "path_with_removed_args": path_with_removed_args,
+                        "named_parameter_values": named_parameter_values,
+                        "canned_query": canned_query,
+                        "success_message": request.args.get("_success") or "",
+                        "canned_write": True,
+                    }
+
+                return (
+                    {
+                        "database": database,
+                        "rows": [],
+                        "truncated": False,
+                        "columns": [],
+                        "query": {"sql": sql, "params": params},
+                    },
+                    extra_template,
+                    templates,
+                )
+        else:  # Not a write
+            results = await self.ds.execute(
+                database, sql, params, truncate=True, **extra_args
+            )
+            columns = [r[0] for r in results.description]
+
         if canned_query:
             templates.insert(
                 0,
diff --git a/datasette/views/table.py b/datasette/views/table.py
index 2e9515c3..79bf8b08 100644
--- a/datasette/views/table.py
+++ b/datasette/views/table.py
@@ -221,6 +221,22 @@ class RowTableShared(DataView):
 class TableView(RowTableShared):
     name = "table"
 
+    async def post(self, request, db_name, table_and_format):
+        # Handle POST to a canned query
+        canned_query = self.ds.get_canned_query(db_name, table_and_format)
+        assert canned_query, "You may only POST to a canned query"
+        return await QueryView(self.ds).data(
+            request,
+            db_name,
+            None,
+            canned_query["sql"],
+            metadata=canned_query,
+            editable=False,
+            canned_query=table_and_format,
+            named_parameters=canned_query.get("params"),
+            write=bool(canned_query.get("write")),
+        )
+
     async def data(
         self,
         request,
@@ -241,6 +257,8 @@ class TableView(RowTableShared):
                 metadata=canned_query,
                 editable=False,
                 canned_query=table,
+                named_parameters=canned_query.get("params"),
+                write=bool(canned_query.get("write")),
             )
 
         db = self.ds.databases[database]
diff --git a/docs/sql_queries.rst b/docs/sql_queries.rst
index c3efd930..dc239a84 100644
--- a/docs/sql_queries.rst
+++ b/docs/sql_queries.rst
@@ -161,11 +161,12 @@ You can set a default fragment hash that will be included in the link to the can
 
     {
         "databases": {
-        "fixtures": {
-            "queries": {
-                "neighborhood_search": {
-                    "sql": "select neighborhood, facet_cities.name, state\nfrom facetable join facet_cities on facetable.city_id = facet_cities.id\nwhere neighborhood like '%' || :text || '%' order by neighborhood;",
-                    "fragment": "fragment-goes-here"
+            "fixtures": {
+                "queries": {
+                    "neighborhood_search": {
+                        "sql": "select neighborhood, facet_cities.name, state\nfrom facetable join facet_cities on facetable.city_id = facet_cities.id\nwhere neighborhood like '%' || :text || '%' order by neighborhood;",
+                        "fragment": "fragment-goes-here"
+                    }
                 }
             }
         }
@@ -173,6 +174,60 @@ You can set a default fragment hash that will be included in the link to the can
 
 `See here <https://latest.datasette.io/fixtures#queries>`__ for a demo of this in action.
 
+.. _canned_queries_writable:
+
+Writable canned queries
+~~~~~~~~~~~~~~~~~~~~~~~
+
+Canned queries by default are read-only. You can use the ``"write": true`` key to indicate that a canned query can write to the database.
+
+.. code-block:: json
+
+    {
+        "databases": {
+            "mydatabase": {
+                "queries": {
+                    "add_name": {
+                        "sql": "INSERT INTO names (name) VALUES (:name)",
+                        "write": true
+                    }
+                }
+            }
+        }
+    }
+
+This configuration will create a page at ``/mydatabase/add_name`` displaying a form with a ``name`` field. Submitting that form will execute the configured ``INSERT`` query.
+
+You can customize how Datasette represents success and errors using the following optional properties:
+
+- ``on_success_message`` - the message shown when a query is successful
+- ``on_success_redirect`` - the path or URL the user is redirected to on success
+- ``on_error_message`` - the message shown when a query throws an error
+- ``on_error_redirect`` - the path or URL the user is redirected to on error
+
+For example:
+
+.. code-block:: json
+
+    {
+        "databases": {
+            "mydatabase": {
+                "queries": {
+                    "add_name": {
+                        "sql": "INSERT INTO names (name) VALUES (:name)",
+                        "write": true,
+                        "on_success_message": "Name inserted",
+                        "on_success_redirect": "/mydatabase/names",
+                        "on_error_message": "Name insert failed",
+                        "on_error_redirect": "/mydatabase"
+                    }
+                }
+            }
+        }
+    }
+
+You may wish to use this feature in conjunction with :ref:`authentication`.
+
 .. _pagination:
 
 Pagination
diff --git a/tests/fixtures.py b/tests/fixtures.py
index daff0168..78a54c68 100644
--- a/tests/fixtures.py
+++ b/tests/fixtures.py
@@ -14,7 +14,7 @@ import string
 import tempfile
 import textwrap
 import time
-from urllib.parse import unquote, quote
+from urllib.parse import unquote, quote, urlencode
 
 
 # This temp file is used by one of the plugin config tests
@@ -54,10 +54,26 @@ class TestClient:
     async def get(
         self, path, allow_redirects=True, redirect_count=0, method="GET", cookies=None
     ):
-        return await self._get(path, allow_redirects, redirect_count, method, cookies)
+        return await self._request(
+            path, allow_redirects, redirect_count, method, cookies
+        )
 
-    async def _get(
-        self, path, allow_redirects=True, redirect_count=0, method="GET", cookies=None
+    @async_to_sync
+    async def post(
+        self, path, post_data=None, allow_redirects=True, redirect_count=0, cookies=None
+    ):
+        return await self._request(
+            path, allow_redirects, redirect_count, "POST", cookies, post_data
+        )
+
+    async def _request(
+        self,
+        path,
+        allow_redirects=True,
+        redirect_count=0,
+        method="GET",
+        cookies=None,
+        post_data=None,
     ):
         query_string = b""
         if "?" in path:
@@ -83,7 +99,13 @@ class TestClient:
             "headers": headers,
         }
         instance = ApplicationCommunicator(self.asgi_app, scope)
-        await instance.send_input({"type": "http.request"})
+
+        if post_data:
+            body = urlencode(post_data, doseq=True).encode("utf-8")
+            await instance.send_input({"type": "http.request", "body": body})
+        else:
+            await instance.send_input({"type": "http.request"})
+
         # First message back should be response.start with headers and status
         messages = []
         start = await instance.receive_output(2)
@@ -110,7 +132,7 @@ class TestClient:
                 redirect_count, self.max_redirects
             )
             location = response.headers["Location"]
-            return await self._get(
+            return await self._request(
                 location, allow_redirects=True, redirect_count=redirect_count + 1
             )
         return response
@@ -128,6 +150,7 @@ def make_app_client(
     inspect_data=None,
     static_mounts=None,
     template_dir=None,
+    metadata=None,
 ):
     with tempfile.TemporaryDirectory() as tmpdir:
         filepath = os.path.join(tmpdir, filename)
@@ -161,7 +184,7 @@ def make_app_client(
             immutables=immutables,
             memory=memory,
             cors=cors,
-            metadata=METADATA,
+            metadata=metadata or METADATA,
             plugins_dir=PLUGINS_DIR,
             config=config,
             inspect_data=inspect_data,
diff --git a/tests/test_canned_write.py b/tests/test_canned_write.py
new file mode 100644
index 00000000..52c8aec2
--- /dev/null
+++ b/tests/test_canned_write.py
@@ -0,0 +1,88 @@
+import pytest
+from .fixtures import make_app_client
+
+
+@pytest.fixture
+def canned_write_client():
+    for client in make_app_client(
+        extra_databases={"data.db": "create table names (name text)"},
+        metadata={
+            "databases": {
+                "data": {
+                    "queries": {
+                        "add_name": {
+                            "sql": "insert into names (name) values (:name)",
+                            "write": True,
+                            "on_success_redirect": "/data/add_name?success",
+                        },
+                        "add_name_specify_id": {
+                            "sql": "insert into names (rowid, name) values (:rowid, :name)",
+                            "write": True,
+                            "on_error_redirect": "/data/add_name_specify_id?error",
+                        },
+                        "delete_name": {
+                            "sql": "delete from names where rowid = :rowid",
+                            "write": True,
+                            "on_success_message": "Name deleted",
+                        },
+                        "update_name": {
+                            "sql": "update names set name = :name where rowid = :rowid",
+                            "params": ["rowid", "name"],
+                            "write": True,
+                        },
+                    }
+                }
+            }
+        },
+    ):
+        yield client
+
+
+def test_insert(canned_write_client):
+    response = canned_write_client.post(
+        "/data/add_name", {"name": "Hello"}, allow_redirects=False
+    )
+    assert 302 == response.status
+    assert "/data/add_name?success" == response.headers["Location"]
+    messages = canned_write_client.ds.unsign(
+        response.cookies["ds_messages"], "messages"
+    )
+    assert [["Query executed, 1 row affected", 1]] == messages
+
+
+def test_custom_success_message(canned_write_client):
+    response = canned_write_client.post(
+        "/data/delete_name", {"rowid": 1}, allow_redirects=False
+    )
+    assert 302 == response.status
+    messages = canned_write_client.ds.unsign(
+        response.cookies["ds_messages"], "messages"
+    )
+    assert [["Name deleted", 1]] == messages
+
+
+def test_insert_error(canned_write_client):
+    canned_write_client.post("/data/add_name", {"name": "Hello"})
+    response = canned_write_client.post(
+        "/data/add_name_specify_id",
+        {"rowid": 1, "name": "Should fail"},
+        allow_redirects=False,
+    )
+    assert 302 == response.status
+    assert "/data/add_name_specify_id?error" == response.headers["Location"]
+    messages = canned_write_client.ds.unsign(
+        response.cookies["ds_messages"], "messages"
+    )
+    assert [["UNIQUE constraint failed: names.rowid", 3]] == messages
+    # How about with a custom error message?
+    canned_write_client.ds._metadata["databases"]["data"]["queries"][
+        "add_name_specify_id"
+    ]["on_error_message"] = "ERROR"
+    response = canned_write_client.post(
+        "/data/add_name_specify_id",
+        {"rowid": 1, "name": "Should fail"},
+        allow_redirects=False,
+    )
+    assert [["ERROR", 3]] == canned_write_client.ds.unsign(
+        response.cookies["ds_messages"], "messages"
+    )