Emit db.query spans from the three write entry points

execute_write(), execute_write_script() and execute_write_many() were the
only Database methods that ran SQL without producing an OpenTelemetry
span, so any instance doing writes - which is every instance, since
Datasette builds its internal catalog through these methods at startup -
showed reads in a trace and nothing else. The same db.system,
db.namespace and db.query.text attributes the read path already sets now
appear here, with db.query.text going through sql_attribute() so
attacker-supplied SQL cannot put an unbounded string on a span.

execute_write_many() records the parameter-set count as
datasette.param_sets, not datasette.rows_returned. executemany() consumes
parameter sets and returns no rows at all, so a rows_returned name would
be describing something that does not exist - and a consumer building a
"rows written" dashboard on top of it would be charting the wrong number.

These spans only cover the event-loop side of a write. The time actually
spent waiting on the write queue and executing on the write thread is not
attributed yet; that needs context propagation across the thread
boundary and lands separately. Writes with block=False are worse still -
execute_write_fn returns before the write happens, so the span closes
early. Span links fix that later.

As with the read path, the existing `with trace(...)` wrappers stay put
and the new spans nest inside them, so ?_trace=1 keeps working
unchanged - including execute_write_many's `count`, which the old tracer
stashes through the context manager's return value.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

datasette/database.py

@@ -259,10 +259,21 @@ class Database:
                 cursor, return_all=return_all, returning_limit=returning_limit
             )
 
-        with trace("sql", database=self.name, sql=sql.strip(), params=params):
-            results = await self.execute_write_fn(
-                _inner, block=block, request=request, transaction=transaction
-            )
+        # SIM117 wants these two context managers merged. They are kept nested
+        # deliberately: the hand-rolled tracer's wrapper is on its way out, and
+        # nesting makes removing it a single-line deletion.
+        with trace(  # noqa: SIM117
+            "sql", database=self.name, sql=sql.strip(), params=params
+        ):
+            with tracer.start_as_current_span("db.query") as span:
+                span.set_attribute("db.system", "sqlite")
+                span.set_attribute("db.namespace", self.name)
+                span.set_attribute("db.query.text", sql_attribute(sql))
+                if params:
+                    span.set_attribute("datasette.param_count", len(params))
+                results = await self.execute_write_fn(
+                    _inner, block=block, request=request, transaction=transaction
+                )
         return results
 
     async def execute_write_script(self, sql, block=True, request=None):
@@ -271,10 +282,18 @@ class Database:
         def _inner(conn):
             return conn.executescript(sql)
 
-        with trace("sql", database=self.name, sql=sql.strip(), executescript=True):
-            results = await self.execute_write_fn(
-                _inner, block=block, transaction=False, request=request
-            )
+        # Nested on purpose - see the note in execute_write().
+        with trace(  # noqa: SIM117
+            "sql", database=self.name, sql=sql.strip(), executescript=True
+        ):
+            with tracer.start_as_current_span("db.query") as span:
+                span.set_attribute("db.system", "sqlite")
+                span.set_attribute("db.namespace", self.name)
+                span.set_attribute("db.query.text", sql_attribute(sql))
+                span.set_attribute("datasette.executescript", True)
+                results = await self.execute_write_fn(
+                    _inner, block=block, transaction=False, request=request
+                )
         return results
 
     async def execute_write_many(self, sql, params_seq, block=True, request=None):
@@ -291,12 +310,21 @@ class Database:
 
             return conn.executemany(sql, count_params(params_seq)), count
 
+        # Nested on purpose - see the note in execute_write().
         with trace(
             "sql", database=self.name, sql=sql.strip(), executemany=True
         ) as kwargs:
-            results, count = await self.execute_write_fn(
-                _inner, block=block, request=request
-            )
+            with tracer.start_as_current_span("db.query") as span:
+                span.set_attribute("db.system", "sqlite")
+                span.set_attribute("db.namespace", self.name)
+                span.set_attribute("db.query.text", sql_attribute(sql))
+                span.set_attribute("datasette.executemany", True)
+                results, count = await self.execute_write_fn(
+                    _inner, block=block, request=request
+                )
+                # count is the number of parameter *sets* consumed by
+                # executemany(), not a row count - executemany returns no rows.
+                span.set_attribute("datasette.param_sets", count)
             kwargs["count"] = count
         return results
 

tests/test_telemetry.py

@@ -6,6 +6,7 @@ import sys
 import pytest
 from opentelemetry.trace import StatusCode
 
+from datasette.app import Datasette
 from datasette.telemetry import MAX_SQL_LENGTH, sql_attribute
 
 SECRET_PARAM_VALUE = "SUPER_SECRET_PARAM_VALUE_XYZ_123"
@@ -17,6 +18,21 @@ def _db_query_spans(otel_spans):
     return [span for span in otel_spans.get_finished_spans() if span.name == "db.query"]
 
 
+def _spans_for_namespace(otel_spans, namespace):
+    """
+    db.query spans belonging to one database.
+
+    Datasette queries its internal catalog constantly - including while a
+    Datasette instance is being constructed - so a test that just grabbed
+    every db.query span would be reading someone else's traffic.
+    """
+    return [
+        span
+        for span in _db_query_spans(otel_spans)
+        if span.attributes["db.namespace"] == namespace
+    ]
+
+
 def _all_attribute_values(otel_spans):
     "Every attribute value across every finished span, for the 'no leaked param values' test."
     values = []
@@ -192,3 +208,62 @@ async def test_suppressed_sql_error_is_not_a_span_error(ds_client, otel_spans):
     assert span.status.status_code == StatusCode.UNSET
     assert span.attributes["datasette.sql_error_suppressed"] is True
     assert not [event for event in span.events if event.name == "exception"]
+
+
+@pytest.mark.asyncio
+async def test_execute_write_produces_db_query_span(otel_spans):
+    # Named in-memory databases are shared-cache, so every test in this file
+    # needs its own name or the second `create table` hits an existing table.
+    db = Datasette(memory=True).add_memory_database("t03_write_span")
+    await db.execute_write("create table docs (id integer primary key, name text)")
+    await db.execute_write("insert into docs (id, name) values (?, ?)", [1, "one"])
+
+    spans = _spans_for_namespace(otel_spans, "t03_write_span")
+    assert spans, "expected db.query spans from execute_write()"
+    span = spans[-1]
+
+    assert span.attributes["db.system"] == "sqlite"
+    assert span.attributes["db.namespace"] == "t03_write_span"
+    assert span.attributes["db.query.text"] == (
+        "insert into docs (id, name) values (?, ?)"
+    )
+    assert span.attributes["datasette.param_count"] == 2
+
+
+@pytest.mark.asyncio
+async def test_execute_write_script_sets_executescript_attribute(otel_spans):
+    db = Datasette(memory=True).add_memory_database("t03_write_script_span")
+    await db.execute_write_script(
+        "create table docs (id integer primary key);\n"
+        "insert into docs (id) values (1);"
+    )
+
+    spans = _spans_for_namespace(otel_spans, "t03_write_script_span")
+    assert spans, "expected a db.query span from execute_write_script()"
+    span = spans[-1]
+
+    assert span.attributes["db.system"] == "sqlite"
+    assert span.attributes["datasette.executescript"] is True
+    assert "insert into docs" in span.attributes["db.query.text"]
+
+
+@pytest.mark.asyncio
+async def test_execute_write_many_records_param_sets_not_rows_returned(otel_spans):
+    db = Datasette(memory=True).add_memory_database("t03_write_many_span")
+    await db.execute_write("create table docs (id integer primary key)")
+    await db.execute_write_many(
+        "insert into docs (id) values (?)", [[i] for i in range(1, 6)]
+    )
+
+    spans = _spans_for_namespace(otel_spans, "t03_write_many_span")
+    many_spans = [
+        span for span in spans if span.attributes.get("datasette.executemany") is True
+    ]
+    assert len(many_spans) == 1
+    span = many_spans[0]
+
+    assert span.attributes["datasette.param_sets"] == 5
+    # executemany() consumes parameter sets and returns no rows at all, so
+    # calling this a row count would be a lie. Asserted explicitly because the
+    # attribute really was named datasette.rows_returned at one point.
+    assert "datasette.rows_returned" not in span.attributes