If just a single insert, link to row page

Refs #2742

datasette/templates/execute_write.html

@@ -90,7 +90,7 @@
 <p>Execute SQL to insert, update or delete rows in this database.</p>
 
 {% if execution_message %}
-    <p class="{% if execution_ok %}message-info{% else %}message-error{% endif %}">{{ execution_message }}</p>
+    <p class="{% if execution_ok %}message-info{% else %}message-error{% endif %}">{{ execution_message }}{% for link in execution_links %} <a href="{{ link.href }}">{{ link.label }}</a>{% endfor %}</p>
 {% endif %}
 
 <form class="sql core" action="{{ urls.database(database) }}/-/execute-write" method="post">

datasette/views/database.py

@@ -18,8 +18,10 @@ from datasette.utils import (
     await_me_maybe,
     call_with_supported_arguments,
     named_parameters as derive_named_parameters,
+    escape_sqlite,
     format_bytes,
     make_slot_function,
+    path_from_row_pks,
     tilde_decode,
     to_css_class,
     validate_sql_select,
@@ -678,6 +680,43 @@ async def _prepare_execute_write(datasette, db, sql, params, actor):
     return parameter_names, params, analysis
 
 
+async def _inserted_row_url(datasette, db, analysis, cursor):
+    if cursor.rowcount != 1:
+        return None
+    lastrowid = getattr(cursor, "lastrowid", None)
+    if lastrowid is None:
+        return None
+    direct_inserts = [
+        access
+        for access in analysis.table_accesses
+        if access.operation == "insert"
+        and access.source is None
+        and access.database == db.name
+    ]
+    if len(direct_inserts) != 1:
+        return None
+    table = direct_inserts[0].table
+    pks = await db.primary_keys(table)
+    use_rowid = not pks
+    select = (
+        "rowid"
+        if use_rowid
+        else ", ".join(escape_sqlite(primary_key) for primary_key in pks)
+    )
+    try:
+        result = await db.execute(
+            "select {} from {} where rowid = ?".format(select, escape_sqlite(table)),
+            [lastrowid],
+        )
+    except sqlite3.DatabaseError:
+        return None
+    row = result.first()
+    if row is None:
+        return None
+    row_path = path_from_row_pks(row, pks, use_rowid)
+    return datasette.urls.row(db.name, table, row_path)
+
+
 def _apply_query_data_types(data):
     typed = dict(data)
     for key in ("hide_sql", "is_published"):
@@ -824,10 +863,12 @@ class ExecuteWriteView(BaseView):
         analysis=None,
         analysis_error=None,
         execution_message=None,
+        execution_links=None,
         execution_ok=None,
         status=200,
     ):
         parameter_values = parameter_values or {}
+        execution_links = execution_links or []
         parameter_names = []
         analysis_rows = []
         table_columns = await _table_columns(self.ds, db.name)
@@ -869,6 +910,7 @@ class ExecuteWriteView(BaseView):
                     row for row in analysis_rows if row["operation"] != "read"
                 ],
                 "execution_message": execution_message,
+                "execution_links": execution_links,
                 "execution_ok": execution_ok,
                 "execute_disabled": bool(
                     (not sql)
@@ -964,6 +1006,12 @@ class ExecuteWriteView(BaseView):
                 )
             )
 
+        inserted_row_url = await _inserted_row_url(self.ds, db, analysis, cursor)
+        execution_links = (
+            [{"href": inserted_row_url, "label": "View row"}]
+            if inserted_row_url
+            else []
+        )
         return await self._render_form(
             request,
             db,
@@ -971,6 +1019,7 @@ class ExecuteWriteView(BaseView):
             parameter_values={name: params.get(name, "") for name in parameter_names},
             analysis=analysis,
             execution_message=message,
+            execution_links=execution_links,
             execution_ok=True,
         )
 

tests/test_queries.py

@@ -849,6 +849,48 @@ async def test_execute_write_post_requires_database_and_table_permissions():
     assert (await db.execute("select name from dogs")).first()[0] == "Cleo"
 
 
+@pytest.mark.asyncio
+async def test_execute_write_insert_links_to_inserted_row():
+    ds = Datasette(memory=True, default_deny=True)
+    ds.root_enabled = True
+    db = ds.add_memory_database("execute_write_insert_link", name="data")
+    await db.execute_write("create table dogs (id integer primary key, name text)")
+    await db.execute_write("create table log (id integer primary key, message text)")
+    await db.execute_write("insert into log (message) values ('existing')")
+    await db.execute_write("""
+        create trigger dogs_after_insert after insert on dogs begin
+            insert into log (message) values (new.name);
+        end
+    """)
+    await ds.invoke_startup()
+
+    insert_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "root"},
+        data={
+            "sql": "insert into dogs (name) values (:name)",
+            "name": "Cleo",
+        },
+    )
+    update_response = await ds.client.post(
+        "/data/-/execute-write",
+        actor={"id": "root"},
+        data={
+            "sql": "update dogs set name = :name where id = :id",
+            "name": "Cleo 2",
+            "id": "1",
+        },
+    )
+
+    assert insert_response.status_code == 200
+    assert "Query executed, 1 row affected" in insert_response.text
+    assert '<a href="/data/dogs/1">View row</a>' in insert_response.text
+    assert "/data/log/2" not in insert_response.text
+    assert update_response.status_code == 200
+    assert "Query executed, 1 row affected" in update_response.text
+    assert "View row" not in update_response.text
+
+
 @pytest.mark.asyncio
 async def test_execute_write_post_rejects_read_only_sql():
     ds = Datasette(memory=True, default_deny=True)