remote/datasette
1
0
Fork 0
mirror of https://github.com/simonw/datasette.git synced 2026-09-02 22:54:08 +02:00
Code Issues Projects Releases Packages Wiki Activity

Merge 1052dc5c7b into bdc9731740

Browse source
This commit is contained in:
Alex Garcia 2026-09-01 23:25:01 +00:00 • committed by GitHub
commit 08aa236fd9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 2552 additions and 137 deletions
Download patch file Download diff file Expand all files Collapse all files

110
datasette/app.py
View file

@ -49,6 +49,8 @@ from .events import Event
from .plugins import DEFAULT_PLUGINS, get_plugins, pm
from .renderer import json_renderer
from .resources import DatabaseResource, TableResource
from .telemetry import tracer
from .telemetry_registry import STARTUP
from .tokens import TokenInvalid
from .tracer import AsgiTracer
from .url_builder import Urls
@ -778,57 +780,69 @@ class Datasette:
# This must be called for Datasette to be in a usable state
if self._startup_invoked:
return
# Register event classes
event_classes = []
for hook in pm.hook.register_events(datasette=self):
extra_classes = await await_me_maybe(hook)
if extra_classes:
event_classes.extend(extra_classes)
self.event_classes = tuple(event_classes)
# invoke_startup() runs before any request exists, so every span its
# children create - the register_* hook dispatches, the internal
# catalog's db.query/db.write spans, and the prepare_connection
# warm-up of the read connections those touch - would otherwise be
# its own orphan root trace: around twenty of them on a fresh
# instance. Bracketing the whole thing gives them somewhere to belong.
# A connection warmed lazily later, by a request touching a new
# database for the first time, nests under that request instead:
# this span has already ended by then.
with tracer.start_as_current_span(STARTUP):
# Register event classes
event_classes = []
for hook in pm.hook.register_events(datasette=self):
extra_classes = await await_me_maybe(hook)
if extra_classes:
event_classes.extend(extra_classes)
self.event_classes = tuple(event_classes)
# Register actions, but watch out for duplicate name/abbr
action_names = {}
action_abbrs = {}
for hook in pm.hook.register_actions(datasette=self):
if hook:
for action in hook:
if (
action.name in action_names
and action != action_names[action.name]
):
raise StartupError(f"Duplicate action name: {action.name}")
if (
action.abbr
and action.abbr in action_abbrs
and action != action_abbrs[action.abbr]
):
raise StartupError(f"Duplicate action abbr: {action.abbr}")
action_names[action.name] = action
if action.abbr:
action_abbrs[action.abbr] = action
self.actions[action.name] = action
# Register actions, but watch out for duplicate name/abbr
action_names = {}
action_abbrs = {}
for hook in pm.hook.register_actions(datasette=self):
if hook:
for action in hook:
if (
action.name in action_names
and action != action_names[action.name]
):
raise StartupError(f"Duplicate action name: {action.name}")
if (
action.abbr
and action.abbr in action_abbrs
and action != action_abbrs[action.abbr]
):
raise StartupError(f"Duplicate action abbr: {action.abbr}")
action_names[action.name] = action
if action.abbr:
action_abbrs[action.abbr] = action
self.actions[action.name] = action
# Register column types (classes, not instances)
self._column_types = {}
for hook in pm.hook.register_column_types(datasette=self):
if hook:
for ct_cls in hook:
if ct_cls.name in self._column_types:
raise StartupError(f"Duplicate column type name: {ct_cls.name}")
self._column_types[ct_cls.name] = ct_cls
# Register column types (classes, not instances)
self._column_types = {}
for hook in pm.hook.register_column_types(datasette=self):
if hook:
for ct_cls in hook:
if ct_cls.name in self._column_types:
raise StartupError(
f"Duplicate column type name: {ct_cls.name}"
)
self._column_types[ct_cls.name] = ct_cls
for hook in pm.hook.prepare_jinja2_environment(
env=self._jinja_env, datasette=self
):
await await_me_maybe(hook)
# Ensure internal tables and metadata are populated before startup hooks
await self._refresh_schemas()
await self._save_queries_from_config()
# Load column_types from config into internal DB
await self._apply_column_types_config()
for hook in pm.hook.startup(datasette=self):
await await_me_maybe(hook)
self._startup_invoked = True
for hook in pm.hook.prepare_jinja2_environment(
env=self._jinja_env, datasette=self
):
await await_me_maybe(hook)
# Ensure internal tables and metadata are populated before startup hooks
await self._refresh_schemas()
await self._save_queries_from_config()
# Load column_types from config into internal DB
await self._apply_column_types_config()
for hook in pm.hook.startup(datasette=self):
await await_me_maybe(hook)
self._startup_invoked = True
def sign(self, value, namespace="default"):
return URLSafeSerializer(self._secret, namespace).dumps(value)

484
datasette/database.py
View file

@ -1,18 +1,45 @@
import asyncio
import atexit
import contextvars
import inspect
import os
import queue
import sys
import tempfile
import threading
import time
import uuid
from collections import namedtuple
from pathlib import Path
import sqlite_utils
from opentelemetry import context as otel_context_api
from opentelemetry.trace import Link, Status, StatusCode, get_current_span
from .inspect import inspect_hash
from .telemetry import sql_attribute, sql_operation_name, tracer
from .telemetry_registry import (
DB_COLLECTION_NAME,
DB_NAMESPACE,
DB_OPERATION_NAME,
DB_QUERY,
DB_QUERY_EXECUTE,
DB_QUERY_TEXT,
DB_SYSTEM,
DB_WRITE_EXECUTE,
DB_WRITE_QUEUE_WAIT,
EXECUTEMANY,
EXECUTESCRIPT,
INTERRUPTED,
ISOLATED_CONNECTION,
PARAM_COUNT,
PARAM_SETS,
ROWS_RETURNED,
SQL_ERROR_SUPPRESSED,
TIME_LIMIT_MS,
TRANSACTION,
TRUNCATED,
)
from .tracer import trace
from .utils import (
call_with_supported_arguments,
@ -257,10 +284,24 @@ 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, kind=DB_QUERY.kind) as span:
span.set_attribute(DB_SYSTEM, "sqlite")
span.set_attribute(DB_NAMESPACE, self.name)
span.set_attribute(DB_QUERY_TEXT, sql_attribute(sql))
operation_name = sql_operation_name(sql)
if operation_name:
span.set_attribute(DB_OPERATION_NAME, operation_name)
if params:
span.set_attribute(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):
@ -269,10 +310,22 @@ 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
):
# No db.operation.name here, deliberately: executescript() runs
# several semicolon-separated statements, and semantic conventions
# say the attribute should not be extracted from query text that
# can hold more than one operation - see sql_operation_name().
with tracer.start_as_current_span(DB_QUERY, kind=DB_QUERY.kind) 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(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):
@ -289,12 +342,26 @@ 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, kind=DB_QUERY.kind) 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(EXECUTEMANY, True)
# A single statement run with many parameter sets, so unlike
# execute_write_script() there is exactly one operation to name.
operation_name = sql_operation_name(sql)
if operation_name:
span.set_attribute(DB_OPERATION_NAME, operation_name)
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(PARAM_SETS, count)
kwargs["count"] = count
return results
@ -321,9 +388,18 @@ class Database:
return _run()
if not write:
# Immutable database - no writes can ever occur, so there is no
# write queue to block; run against a fresh read-only connection
# write queue to block; run against a fresh read-only connection.
# A fresh copy_context() is required per submit (not one shared
# copy reused across calls): concurrent execution of the same
# Context raises "RuntimeError: cannot enter context ... already
# entered". This propagates the caller's otel context (e.g. the
# enclosing db.query span) onto the worker thread.
#
# It also propagates every *other* ContextVar - see the note in
# execute_fn() for why that is safe.
ctx = contextvars.copy_context()
return await asyncio.get_running_loop().run_in_executor(
self.ds.executor, _run
self.ds.executor, ctx.run, _run
)
# Threaded mode - send to write thread
return await self._send_to_write_thread(fn, isolated_connection=True)
@ -428,8 +504,24 @@ class Database:
task_id = uuid.uuid5(uuid.NAMESPACE_DNS, "datasette.io")
loop = asyncio.get_running_loop()
reply_future = loop.create_future()
# Captured here, on the event loop, at enqueue time: the otel
# Context (carrying the enclosing db.query span, if any) and the
# timestamp used to build the db.write.queue_wait span once this
# task is dequeued on the write thread. `block` travels with the
# task too, because it decides whether that context is this task's
# parent or only a link target - see `_execute_writes`.
self._write_queue.put(
WriteTask(fn, task_id, loop, reply_future, isolated_connection, transaction)
WriteTask(
fn,
task_id,
loop,
reply_future,
isolated_connection,
transaction,
otel_context_api.get_current(),
time.time_ns(),
block,
)
)
if block:
return await reply_future
@ -443,6 +535,16 @@ class Database:
conn = None
try:
conn = self.connect(write=True)
# This warm-up runs before any write has ever been queued, so
# there is no captured caller context to attach - and a raw
# threading.Thread does not inherit the context of whoever started
# it. Spans created by plugin hooks here are therefore roots even
# when the write thread is started from inside invoke_startup():
# its datasette.startup span is current on the event loop but does
# not cross this thread boundary. Read connections differ - they
# warm up inside executor tasks submitted with copy_context(), so
# their prepare_connection spans do nest under whoever triggered
# them.
self.ds._prepare_connection(conn, self.name)
except Exception as e: # noqa: BLE001
# Stored and re-raised to whoever queues the next write
@ -457,40 +559,119 @@ class Database:
# Best-effort close as the write thread exits
pass
return
exception = None
result = None
if conn_exception is not None:
exception = conn_exception
elif task.isolated_connection:
try:
isolated_connection = self.connect(write=True)
try:
result = task.fn(isolated_connection)
finally:
isolated_connection.close()
try:
self._all_file_connections.remove(isolated_connection)
except ValueError:
# Was probably a memory connection
pass
except Exception as e: # noqa: BLE001
# Write thread must survive any task failure or the database wedges
sys.stderr.write(f"{e}\n")
sys.stderr.flush()
exception = e
# `task.block` decides how this task's spans relate to the
# context captured at enqueue time:
#
# - block=True: the caller genuinely awaits the reply, so
# containment is accurate. Restore that context as current
# (attach below) so db.write.queue_wait/db.write.execute parent
# normally to the request that queued them. The token must be
# detached below in `finally` - a leaked token silently
# poisons this thread's ambient context for every write
# processed after it, and a *wrong*-token detach only logs a
# warning rather than raising, so this pairing is load-bearing
# and easy to get wrong silently.
# - block=False: the caller returned already without awaiting,
# so the enqueueing span may already have closed (and
# exported) before this task's spans even start - parenting to
# it would make a child appear to outlive its already-closed
# parent, which OTel allows but which renders badly in most
# trace UIs. The enqueueing request *caused* this write
# without *containing* it, so nothing is attached here -
# instead each write span is started as its own root (explicit
# empty `context=`, so the write thread's ambient context
# cannot supply a parent either) carrying one `Link` back to
# the enqueueing span's context, built once into
# `write_span_kwargs` and spread into every start_span call
# below.
token = None
write_span_kwargs = {}
if task.block:
token = otel_context_api.attach(task.otel_context)
else:
try:
if task.transaction:
with conn:
conn.execute("BEGIN IMMEDIATE")
result = task.fn(conn)
else:
result = task.fn(conn)
except Exception as e: # noqa: BLE001
sys.stderr.write(f"{e}\n")
sys.stderr.flush()
exception = e
_deliver_write_result(task, result, exception)
enqueueing_span_context = get_current_span(
task.otel_context
).get_span_context()
# No attributes on the link: there is only one kind of link
# here, so naming the relationship would be a constant that
# carries no information a consumer does not already have
# from the link's existence.
links = (
[Link(enqueueing_span_context)]
if enqueueing_span_context.is_valid
else []
)
write_span_kwargs = {
"context": otel_context_api.Context(),
"links": links,
}
try:
exception = None
result = None
# Explicit start_time/end_time rather than a `with` block:
# this span's duration is the time the task actually spent
# waiting in the queue (enqueue -> dequeue), not the near-
# zero time spent constructing/ending the span object here.
tracer.start_span(
DB_WRITE_QUEUE_WAIT,
start_time=task.enqueued_at_ns,
**write_span_kwargs,
).end(end_time=time.time_ns())
if conn_exception is not None:
# fn never runs in this branch, so there is nothing to
# wrap in a db.write.execute span.
exception = conn_exception
elif task.isolated_connection:
try:
with tracer.start_as_current_span(
DB_WRITE_EXECUTE, **write_span_kwargs
) as span:
span.set_attribute(
ISOLATED_CONNECTION,
task.isolated_connection,
)
span.set_attribute(TRANSACTION, task.transaction)
isolated_connection = self.connect(write=True)
try:
result = task.fn(isolated_connection)
finally:
isolated_connection.close()
try:
self._all_file_connections.remove(
isolated_connection
)
except ValueError:
# Was probably a memory connection
pass
except Exception as e: # noqa: BLE001
# Write thread must survive any task failure or the database wedges
sys.stderr.write(f"{e}\n")
sys.stderr.flush()
exception = e
else:
try:
with tracer.start_as_current_span(
DB_WRITE_EXECUTE, **write_span_kwargs
) as span:
span.set_attribute(
ISOLATED_CONNECTION,
task.isolated_connection,
)
span.set_attribute(TRANSACTION, task.transaction)
if task.transaction:
with conn:
conn.execute("BEGIN IMMEDIATE")
result = task.fn(conn)
else:
result = task.fn(conn)
except Exception as e: # noqa: BLE001
sys.stderr.write(f"{e}\n")
sys.stderr.flush()
exception = e
_deliver_write_result(task, result, exception)
finally:
if token is not None:
otel_context_api.detach(token)
async def execute_fn(self, fn):
self._check_not_closed()
@ -512,7 +693,29 @@ class Database:
with self._pending_execute_futures_lock:
self._check_not_closed()
future = self.ds.executor.submit(in_thread)
# A fresh copy_context() is required per submit (not one shared
# copy reused across calls): concurrent execution of the same
# Context raises "RuntimeError: cannot enter context ...
# already entered". This propagates the caller's otel context
# (e.g. the enclosing db.query span) onto the worker thread.
#
# copy_context() is not selective: it also carries Datasette's own
# ContextVars - _skip_permission_checks and _permission_check_cache
# (datasette/permissions.py), _in_datasette_client (app.py) and,
# until the hand-rolled tracer goes, trace_task_id (tracer.py) -
# into worker threads, where they previously took their defaults.
# That is safe, for two reasons. Nothing reads them on a worker
# thread: the permission code that reads the first two is async and
# only ever runs on the event loop. And Context.run() restores the
# thread's previous context when the callable returns, so a value
# cannot outlive the submit that carried it and reach the next task
# on this shared pool - "skip permission checks" in particular can
# never bleed from one request into another's query. Where a value
# would be read - a plugin calling datasette.in_client() or trace()
# from inside an execute_fn callable - seeing the submitting
# request's value is the more accurate answer, not a leak.
ctx = contextvars.copy_context()
future = self.ds.executor.submit(ctx.run, in_thread)
self._pending_execute_futures.add(future)
future.add_done_callback(self._remove_pending_execute_future)
return await asyncio.wrap_future(future)
@ -525,48 +728,149 @@ class Database:
custom_time_limit=None,
page_size=None,
log_sql_errors=True,
table=None,
):
"""Executes sql against db_name in a thread"""
"""Executes sql against db_name in a thread
`table`, if passed, is recorded as the `db.collection.name` span
attribute. It exists for callers that already know which table the
query targets - the table and row views - and is never derived from
`sql` itself: deriving it would be a parse, and on an instance where
anyone can create a table the resulting value set has no ceiling.
"""
self._check_not_closed()
page_size = page_size or self.ds.page_size
time_limit_ms = self.ds.sql_time_limit_ms
# A caller that hands in a budget shorter than the instance-wide
# sql_time_limit_ms is saying "this may not finish, and that is an
# answer I can use" - and every such caller in core does treat the
# timeout as normal: table_counts() stores None per table, facet
# suggestion moves on to the next column, autocomplete falls back to a
# prefix query. Those timeouts are therefore not span errors. Without
# this, the homepage alone emits one red span per table (it counts
# every table under a 10ms budget) on every single hit.
#
# A query that runs out the instance-wide limit is a different event -
# nobody asked for a short budget, so it stays an error.
timeout_expected = bool(custom_time_limit) and custom_time_limit < time_limit_ms
if timeout_expected:
time_limit_ms = custom_time_limit
def sql_operation_in_thread(conn):
time_limit_ms = self.ds.sql_time_limit_ms
if custom_time_limit and custom_time_limit < time_limit_ms:
time_limit_ms = custom_time_limit
with sqlite_timelimit(conn, time_limit_ms):
# This span is created inside the worker thread. Its parent is
# resolved from the ambient otel context, which was propagated
# onto this thread via copy_context() at the executor.submit()
# boundary in execute_fn() (or run_in_executor() for immutable
# databases) - so it parents correctly to the enclosing
# db.query span despite running on a different thread.
#
# Exception handling is explicit rather than left to the context
# manager's flags, which apply to every exception type alike. This
# span needs to tell two apart: an expected timeout is never an
# error, while a genuine SQL failure is one unless the caller
# passed log_sql_errors=False, meaning it was probing and treats
# failure as an expected answer. Without the latter, facet
# suggestion marks two spans per text column as failed on every
# table page; without the former, so does every homepage hit.
with tracer.start_as_current_span(
DB_QUERY_EXECUTE,
record_exception=False,
set_status_on_exception=False,
) as execute_span:
try:
cursor = conn.cursor()
cursor.execute(sql, params if params is not None else {})
max_returned_rows = self.ds.max_returned_rows
if max_returned_rows == page_size:
max_returned_rows += 1
if max_returned_rows and truncate:
rows = cursor.fetchmany(max_returned_rows + 1)
truncated = len(rows) > max_returned_rows
rows = rows[:max_returned_rows]
else:
rows = cursor.fetchall()
truncated = False
except (sqlite3.OperationalError, sqlite3.DatabaseError) as e:
if e.args == ("interrupted",):
raise QueryInterrupted(e, sql, params)
with sqlite_timelimit(conn, time_limit_ms):
try:
cursor = conn.cursor()
cursor.execute(sql, params if params is not None else {})
max_returned_rows = self.ds.max_returned_rows
if max_returned_rows == page_size:
max_returned_rows += 1
if max_returned_rows and truncate:
rows = cursor.fetchmany(max_returned_rows + 1)
truncated = len(rows) > max_returned_rows
rows = rows[:max_returned_rows]
else:
rows = cursor.fetchall()
truncated = False
except (sqlite3.OperationalError, sqlite3.DatabaseError) as e:
if e.args == ("interrupted",):
raise QueryInterrupted(e, sql, params)
if log_sql_errors:
sys.stderr.write(
f"ERROR: conn={conn}, sql = {sql!r}, params = {params}: {e}\n"
)
sys.stderr.flush()
raise
except QueryInterrupted as e:
if not timeout_expected:
execute_span.record_exception(e)
execute_span.set_status(Status(StatusCode.ERROR, str(e)))
raise
except Exception as e:
if log_sql_errors:
sys.stderr.write(
f"ERROR: conn={conn}, sql = {sql!r}, params = {params}: {e}\n"
)
sys.stderr.flush()
execute_span.record_exception(e)
execute_span.set_status(Status(StatusCode.ERROR, str(e)))
raise
if truncate:
return Results(rows, truncated, cursor.description)
if truncate:
return Results(rows, truncated, cursor.description)
else:
return Results(rows, False, cursor.description)
else:
return Results(rows, False, cursor.description)
with trace("sql", database=self.name, sql=sql.strip(), params=params):
results = await self.execute_fn(sql_operation_in_thread)
# 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
):
# Exception handling is explicit rather than left to the context
# manager's defaults, so that callers passing log_sql_errors=False
# can be honoured - see the comment on the generic handler below.
with tracer.start_as_current_span(
DB_QUERY,
kind=DB_QUERY.kind,
record_exception=False,
set_status_on_exception=False,
) 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(TIME_LIMIT_MS, time_limit_ms)
operation_name = sql_operation_name(sql)
if operation_name:
span.set_attribute(DB_OPERATION_NAME, operation_name)
if table:
span.set_attribute(DB_COLLECTION_NAME, table)
if params:
span.set_attribute(PARAM_COUNT, len(params))
try:
results = await self.execute_fn(sql_operation_in_thread)
except QueryInterrupted as e:
# datasette.interrupted is set either way - it is the
# signal worth having. Only the ERROR status is
# conditional; see the timeout_expected comment above.
span.set_attribute(INTERRUPTED, True)
if not timeout_expected:
span.set_status(Status(StatusCode.ERROR, str(e)))
span.record_exception(e)
raise
except Exception as e:
# log_sql_errors=False means the caller is probing and
# treats failure as an expected answer, not an error.
# Facet suggestion is the big one: it runs json_type()
# against every column precisely to find out which ones
# raise, so a table with N text columns would otherwise
# mark N queries per page as failed - burying real errors
# and setting off any alerting based on span status.
if log_sql_errors:
span.record_exception(e)
span.set_status(Status(StatusCode.ERROR, str(e)))
else:
span.set_attribute(SQL_ERROR_SUPPRESSED, True)
raise
span.set_attribute(TRUNCATED, results.truncated)
span.set_attribute(ROWS_RETURNED, len(results.rows))
return results
@property
@ -854,16 +1158,28 @@ def _apply_write_wrapper(fn, wrapper_factory, track_event):
class WriteTask:
__slots__ = (
"block",
"enqueued_at_ns",
"fn",
"isolated_connection",
"loop",
"otel_context",
"reply_future",
"task_id",
"transaction",
)
def __init__(
self, fn, task_id, loop, reply_future, isolated_connection, transaction
self,
fn,
task_id,
loop,
reply_future,
isolated_connection,
transaction,
otel_context,
enqueued_at_ns,
block,
):
self.fn = fn
self.task_id = task_id
@ -871,6 +1187,14 @@ class WriteTask:
self.reply_future = reply_future
self.isolated_connection = isolated_connection
self.transaction = transaction
self.otel_context = otel_context
self.enqueued_at_ns = enqueued_at_ns
# Whether the enqueueing caller awaits the reply future. Decides how
# `_execute_writes` relates this task's spans to `otel_context`:
# parent (block=True) or span-link target (block=False). See the
# comment at the WriteTask construction site in
# `_send_to_write_thread`.
self.block = block
def _deliver_write_result(task, result, exception):

117
datasette/telemetry.py Normal file
View file

@ -0,0 +1,117 @@
"""
OpenTelemetry integration for Datasette core.
Core depends on `opentelemetry-api` only. It never creates a
`TracerProvider`, never configures an exporter, and never touches
sampling - that is the responsibility of whoever is running Datasette
(an `opentelemetry-instrument` agent, a future plugin, or a test
harness).
With no provider installed every span produced here is a
`NonRecordingSpan`. That is not free - a table page emits ~58 spans -
but it is below what an end-to-end page benchmark can resolve: measured
across 15 runs of a 5,000-row table page, the median moved 9.80ms to
9.98ms while run-to-run spread was 1.4ms. Installing an SDK provider is
what costs something measurable.
"""
import re
from opentelemetry import trace as otel_trace
from .version import __version__
# The semantic-convention version whose spellings this instrumentation
# actually emits. Deliberately NOT the latest release.
#
# A schema URL is a machine-readable claim: a consumer doing schema
# translation replays the renames between the declared version and the one
# it wants, so the claim has to name the version whose spellings are on the
# wire. A wrong one makes translation wrong rather than merely uninformative.
#
# Datasette emits `db.system`, which was renamed to `db.system.name` in
# semconv 1.30.0. Everything else it emits (`db.namespace`, `db.query.text`,
# `db.operation.name`, `db.collection.name`) has been current since 1.26.0.
# So 1.29.0 is the highest version at which every name emitted here is the
# current spelling. Everything under `datasette.*` is Datasette's own and
# outside semconv, so it is unaffected either way.
#
# Declaring 1.43.0 would be false about `db.system`, and would actively STOP
# a consumer translating it forward, because it asserts the rename already
# happened. Bump this deliberately, in the same commit as the attribute
# renames it implies - it is a claim about the names, not decoration.
SCHEMA_URL = "https://opentelemetry.io/schemas/1.29.0"
tracer = otel_trace.get_tracer("datasette", __version__, schema_url=SCHEMA_URL)
MAX_SQL_LENGTH = 2048
def sql_attribute(sql: str) -> str:
"Truncate SQL text so it is safe to attach to a span as an attribute."
sql = sql.strip()
if len(sql) <= MAX_SQL_LENGTH:
return sql
return sql[:MAX_SQL_LENGTH] + "…[truncated]"
# db.operation.name is the leading keyword of a statement matched against a
# fixed allowlist - deliberately not a parse.
#
# This runs against arbitrary user-supplied SQL (the `?sql=` query string,
# canned queries, anything typed into the query editor), and the attribute is
# a candidate dimension on a query-duration metric in a later phase. A metric
# series is keyed by its attribute values, so echoing back an arbitrary first
# token would let one visitor's typo mint a new, permanent series. The
# allowlist bounds that at a fixed, small set regardless of what anyone sends.
DB_OPERATION_ALLOWLIST = frozenset(
{
"SELECT",
"INSERT",
"UPDATE",
"DELETE",
"CREATE",
"DROP",
"ALTER",
"PRAGMA",
"EXPLAIN",
"REPLACE",
"VACUUM",
"ANALYZE",
"WITH",
}
)
_LEADING_KEYWORD = re.compile(r"^\s*([A-Za-z]+)")
def sql_operation_name(sql: str) -> str | None:
"""
The statement's leading keyword, if it is one we recognise.
Returns None - never a guess - for anything not on the allowlist,
including a statement that opens with a comment or with punctuation such
as the "(" of a parenthesised SELECT.
Known limitation: a statement beginning with a CTE reports `WITH` rather
than the operation inside it, and a substantial share of Datasette's own
reads take that form. Extracting more than the leading keyword means
handling comment stripping, parenthesised `(SELECT ...) UNION` and
compound names like `CREATE TABLE` - each a special case a hand-rolled
matcher would accrete and eventually get wrong. Omitting a name beats
guessing at one.
Only safe to call with a single statement: `execute_write_script()` runs
several separated by semicolons, and semantic conventions say
`db.operation.name` "SHOULD NOT be extracted from db.query.text, when the
database system supports query text with multiple operations in non-batch
operations" - so that call site does not use this at all rather than
reporting only the first statement's operation.
"""
match = _LEADING_KEYWORD.match(sql)
if not match:
return None
keyword = match.group(1).upper()
if keyword in DB_OPERATION_ALLOWLIST:
return keyword
return None

273
datasette/telemetry_registry.py Normal file
View file

@ -0,0 +1,273 @@
"""
The single source of truth for every span and span attribute that Datasette
core emits.
Three things read this module, which is the point of it existing:
1. **The instrumentation itself.** `Attribute` and `SpanName` subclass `str`,
so a registry entry *is* the string OpenTelemetry wants. Call sites pass
`DB_NAMESPACE` where they used to pass `"db.namespace"` - no wrapper API
over the OTel calls, no parallel structure to keep in step, and a typo is
now an `ImportError` instead of a silently misnamed attribute.
2. **The documentation.** `docs/telemetry_doc.py` renders the span reference
in `docs/internals.rst` from these definitions using cog, and
`cog --check` runs in CI - so the docs cannot drift from the code.
3. **A conformance test.** `tests/test_telemetry_registry.py` makes real
requests, collects every span and attribute actually emitted, and compares
both directions: emitted-but-unregistered catches instrumentation added
without documentation, registered-but-never-emitted catches documentation
describing something that no longer exists. Neither the type system nor
the generated docs can catch that second case.
"""
from opentelemetry.trace import SpanKind
class Attribute(str):
"""
A span attribute key, carrying its own documentation.
Subclasses `str` so it can be handed straight to `set_attribute()`.
"""
__slots__ = ("description", "optional")
def __new__(cls, name, description, optional=False):
self = super().__new__(cls, name)
self.description = description
self.optional = optional
return self
def __repr__(self):
return f"Attribute({str(self)!r})"
class SpanName(str):
"A span name, carrying its documentation and the attributes it may set."
__slots__ = ("attributes", "description", "kind", "prefix")
def __new__(
cls, name, description, attributes=(), prefix=False, kind=SpanKind.INTERNAL
):
self = super().__new__(cls, name)
self.description = description
self.attributes = tuple(attributes)
# True for a span family whose emitted names carry a variable suffix,
# so the conformance test matches by prefix rather than equality.
# Nothing sets it yet.
self.prefix = prefix
# SpanKind.INTERNAL by default - every span Datasette emits describes
# its own internal work. db.query is the one exception: it is a real
# database call, so semantic conventions (and trace UIs, which key
# their database styling off this) expect SpanKind.CLIENT.
self.kind = kind
return self
def __repr__(self):
return f"SpanName({str(self)!r})"
# --- Attributes -----------------------------------------------------------
#
# Shared attributes are defined once and referenced by every span that sets
# them, so "which spans carry db.namespace?" is answerable by grep.
DB_SYSTEM = Attribute("db.system", "Always ``sqlite``.")
DB_NAMESPACE = Attribute("db.namespace", "Name of the database being queried.")
DB_QUERY_TEXT = Attribute(
"db.query.text",
"The SQL, truncated to 2048 characters. Never the parameter values.",
)
DB_OPERATION_NAME = Attribute(
"db.operation.name",
"The statement's leading keyword - ``SELECT``, ``INSERT``, ``CREATE``, and "
"so on - matched against a small fixed allowlist. Omitted rather than set "
"to an arbitrary value: the allowlist exists because this attribute is a "
"candidate dimension for a query-duration metric in a later phase, and "
"echoing an unrecognised first token from user-supplied SQL would be an "
"unbounded-cardinality hazard. Also omitted for "
"``execute_write_script()``, which runs multiple statements - per "
"semantic conventions, the operation name should not be extracted from "
"query text that can contain more than one operation. Note that a "
"statement beginning with a CTE reports ``WITH``, not the operation "
"inside it - a substantial share of Datasette's own reads take that "
"form. Resolving it further would mean parsing.",
optional=True,
)
DB_COLLECTION_NAME = Attribute(
"db.collection.name",
"The primary table, set only where the view already knows it - the table "
"and row pages. Omitted for arbitrary ``?sql=`` queries, where determining "
"the table would mean parsing the query.",
optional=True,
)
PARAM_COUNT = Attribute(
"datasette.param_count",
"Number of bound parameters. Recorded instead of the values themselves.",
optional=True,
)
PARAM_SETS = Attribute(
"datasette.param_sets",
"Number of parameter sets consumed by ``execute_write_many()``. Not a row "
"count - ``executemany()`` returns no rows. The parameter values "
"themselves are never recorded: that sequence can hold thousands of rows.",
optional=True,
)
TIME_LIMIT_MS = Attribute(
"datasette.time_limit_ms",
"The :ref:`setting_sql_time_limit_ms` value this query ran under. Set on "
"reads, which are the queries that time limit applies to.",
optional=True,
)
ROWS_RETURNED = Attribute(
"datasette.rows_returned",
"Number of rows a read returned. Set on the read path only, and only when "
"the read succeeded.",
optional=True,
)
TRUNCATED = Attribute(
"datasette.truncated",
"True if the result was cut short by :ref:`setting_max_returned_rows`.",
optional=True,
)
INTERRUPTED = Attribute(
"datasette.interrupted",
"True if the query was cancelled for exceeding the time limit. The span "
"status is also set to ``ERROR``, unless the caller asked for a budget "
"shorter than :ref:`setting_sql_time_limit_ms` - as table counts, facet "
"suggestion and autocomplete all do - in which case running out of time "
"is an expected answer rather than a failure and the status is left "
"unset.",
optional=True,
)
SQL_ERROR_SUPPRESSED = Attribute(
"datasette.sql_error_suppressed",
"True when the query failed but the caller passed ``log_sql_errors=False``, "
"meaning it was probing and treats failure as an expected answer. Facet "
"suggestion does this against every column.",
optional=True,
)
EXECUTESCRIPT = Attribute(
"datasette.executescript",
"True for ``execute_write_script()``, which runs multiple statements.",
optional=True,
)
EXECUTEMANY = Attribute(
"datasette.executemany",
"True for ``execute_write_many()``, which runs one statement against many "
"parameter sets.",
optional=True,
)
ISOLATED_CONNECTION = Attribute(
"datasette.isolated_connection",
"True if the write ran on its own connection rather than the shared write "
"connection.",
)
TRANSACTION = Attribute(
"datasette.transaction",
"False for statements such as ``VACUUM`` that cannot run inside a transaction.",
)
# --- Spans ----------------------------------------------------------------
DB_QUERY = SpanName(
"db.query",
"A SQL operation issued by Datasette, covering the full round trip "
"including any time spent queued for a thread.",
(
DB_SYSTEM,
DB_NAMESPACE,
DB_QUERY_TEXT,
DB_OPERATION_NAME,
DB_COLLECTION_NAME,
PARAM_COUNT,
PARAM_SETS,
TIME_LIMIT_MS,
ROWS_RETURNED,
TRUNCATED,
INTERRUPTED,
SQL_ERROR_SUPPRESSED,
EXECUTESCRIPT,
EXECUTEMANY,
),
kind=SpanKind.CLIENT,
)
DB_QUERY_EXECUTE = SpanName(
"db.query.execute",
"The read executing inside a SQL worker thread. Child of ``db.query``; the "
"gap between the two is time spent waiting for a thread.",
)
DB_WRITE_QUEUE_WAIT = SpanName(
"db.write.queue_wait",
"Time a write spent waiting in its database's write queue before the write "
"thread picked it up. Child of ``db.query`` for a ``block=True`` write, "
"where the caller awaits the write and containment is accurate. For a "
"``block=False`` write the caller does not await it - the enqueueing "
"request *caused* the write without *containing* it, and the write's "
"spans can outlive the request's own - so this is a root span instead, "
"carrying an OpenTelemetry link back to the enqueueing span rather than "
"a parent. A link records causation without asserting containment, which "
"is exactly the distinction here.",
)
DB_WRITE_EXECUTE = SpanName(
"db.write.execute",
"The write executing on the write thread. Child of ``db.query`` for a "
"``block=True`` write; for ``block=False`` a root span with a link back "
"to the enqueueing span instead - see ``db.write.queue_wait`` above.",
(ISOLATED_CONNECTION, TRANSACTION),
)
STARTUP = SpanName(
"datasette.startup",
"``invoke_startup()`` running: ``register_events``, ``register_actions``, "
"``register_column_types``, ``prepare_jinja2_environment``, internal-database "
"schema catalog refresh (including the ``prepare_connection`` warm-up this "
"triggers for each database touched for the first time), saved queries, "
"column type config and the ``startup`` hook. Runs once per process, before "
"any request exists, so without this span every child it creates would be "
"its own orphan root trace. A connection warmed later - lazily, the first "
"time a *request* touches a new database or thread - nests under that "
"request's own span instead, not under this one, since this span has "
"already ended by then.",
)
SPANS = (
DB_QUERY,
DB_QUERY_EXECUTE,
DB_WRITE_QUEUE_WAIT,
DB_WRITE_EXECUTE,
STARTUP,
)
def span_for(emitted_name):
"""
Resolve an emitted span name to its registry entry, or None.
Handles span families whose emitted names carry a suffix that is not
knowable in advance - `prefix=True` entries. Phase 1 has none, but the
lookup is what the conformance test calls, so it lives here rather than
in the test.
"""
for span in SPANS:
if span.prefix:
if emitted_name.startswith(span):
return span
elif emitted_name == span:
return span
return None
def attribute_allowed(span, emitted_key):
"Whether `emitted_key` is a registered attribute of `span`."
if span is None:
return False
return emitted_key in span.attributes

9
datasette/views/row.py
View file

@ -407,7 +407,7 @@ class RowView(BaseView):
raise Forbidden("You do not have permission to view this table")
results = await resolved.db.execute(
resolved.sql, resolved.params, truncate=True
resolved.sql, resolved.params, truncate=True, table=table
)
columns = [r[0] for r in results.description]
rows = list(results.rows)
@ -652,6 +652,9 @@ class RowView(BaseView):
]
)
try:
# No table= here: this counts incoming references across every
# foreign key pointing at this row, so it spans many tables and
# there is no single value db.collection.name could take.
rows = list(await db.execute(sql, {"id": pk_values[0]}))
except QueryInterrupted:
# Almost certainly hit the timeout
@ -840,7 +843,7 @@ class RowUpdateView(BaseView):
returned_row = None
if data.get("return"):
results = await resolved.db.execute(
resolved.sql, resolved.params, truncate=True
resolved.sql, resolved.params, truncate=True, table=resolved.table
)
returned_row = results.dicts()[0]
result["rows"] = [returned_row]
@ -858,7 +861,7 @@ class RowUpdateView(BaseView):
message_row = returned_row
if message_row is None:
results = await resolved.db.execute(
resolved.sql, resolved.params, truncate=True
resolved.sql, resolved.params, truncate=True, table=resolved.table
)
message_row = results.first()
self.ds.add_message(

16
datasette/views/table.py
View file

@ -1170,6 +1170,7 @@ class TableInsertView(BaseView):
"rowid, " if pks == ["rowid"] else "", table_name, where_clause
),
args,
table=table_name,
)
result["rows"] = fetched_rows.dicts()
else:
@ -1382,7 +1383,9 @@ class TableDropView(BaseView):
"database": database_name,
"table": table_name,
"row_count": (
await db.execute(f"select count(*) from [{table_name}]")
await db.execute(
f"select count(*) from [{table_name}]", table=table_name
)
).single_value(),
"message": 'Pass "confirm": true to confirm',
},
@ -1576,7 +1579,10 @@ class TableAutocompleteView(BaseView):
try:
results = await db.execute(
sql, params, custom_time_limit=AUTOCOMPLETE_TIME_LIMIT_MS
sql,
params,
custom_time_limit=AUTOCOMPLETE_TIME_LIMIT_MS,
table=table_name,
)
except QueryInterrupted:
fallback_where = _autocomplete_prefix_like(pks[0])
@ -1597,6 +1603,7 @@ class TableAutocompleteView(BaseView):
fallback_sql,
params,
custom_time_limit=AUTOCOMPLETE_TIME_LIMIT_MS,
table=table_name,
)
except QueryInterrupted:
return Response.json({"ok": True, "rows": []})
@ -2163,7 +2170,9 @@ async def table_view_data(
# Execute the main query!
try:
results = await db.execute(sql, params, truncate=True, **extra_args)
results = await db.execute(
sql, params, truncate=True, table=table_name, **extra_args
)
except (sqlite3.OperationalError, InvalidSql) as e:
raise DatasetteError(str(e), title="Invalid SQL", status=400)
@ -2439,6 +2448,7 @@ async def _next_value_and_url(
await db.execute(
prefix_lookup_sql,
{**{f"pk{i}": rows[-2][pk] for i, pk in enumerate(pks)}},
table=table_name,
)
).single_value()
if isinstance(prefix, dict) and "value" in prefix:

10
docs/changelog.rst
View file

@ -4,6 +4,16 @@
Changelog
=========
.. _v_unreleased:
Unreleased
----------
- Datasette's database layer now emits `OpenTelemetry <https://opentelemetry.io/>`__ spans: one per query, covering the full round trip including time spent waiting for a SQL worker thread, plus separate child spans for the execution itself and for time spent in the write queue. Datasette core depends on ``opentelemetry-api`` only and never installs an SDK provider, an exporter or a sampler, so there is no effect and no measurable overhead unless tracing is switched on externally - normally with the standard ``opentelemetry-instrument`` agent. See :ref:`internals_telemetry`. (:issue:`1730`)
- :ref:`db.execute(sql, ..., table=None) <database_execute>` has a new optional ``table=`` parameter, naming the table a query is about so it can be recorded on that query's OpenTelemetry span. It has no effect on query execution, and Datasette never derives it from the SQL. (:issue:`1730`)
Nothing is removed by this change: the ``?_trace=1`` query string parameter, the ``trace_debug`` setting and the :ref:`internals_tracer` module all continue to work as before.
.. _v1_0_a38:
1.0a38 (2026-08-06)

130
docs/internals.rst
View file

@ -1963,6 +1963,9 @@ Executes a SQL query against the database and returns the resulting rows (see :r
``log_sql_errors`` - boolean
Should any SQL errors be logged to the console in addition to being raised as an error? Defaults to ``True``.
``table`` - string
The name of the table this query is about, if the caller already knows it. This has no effect on how the query executes - it is recorded as the ``db.collection.name`` attribute on the :ref:`OpenTelemetry span <internals_telemetry>` for the query. Datasette never derives this from the SQL, so leave it unset for queries that do not have one obvious table.
.. _database_results:
Results
@ -2313,6 +2316,133 @@ The ``Database`` class also provides properties and methods for introspecting th
}
}
.. _internals_telemetry:
OpenTelemetry
=============
Datasette core depends on `opentelemetry-api <https://pypi.org/project/opentelemetry-api/>`__ only. It never creates a ``TracerProvider``, never configures an exporter and never sets a sampler. With no OpenTelemetry SDK provider installed, every span described below is a no-op ``NonRecordingSpan``: nothing is recorded, nothing is exported, and the cost does not show up in page latency. Benchmarking a table page with and without this instrumentation, the median moved by less than the run-to-run variation of the benchmark itself.
Turning tracing on is entirely an operational decision made outside of Datasette itself: run Datasette under the standard ``opentelemetry-instrument`` agent, or embed Datasette inside a host application that installs its own provider.
Everything Datasette emits carries the instrumentation scope ``datasette``, versioned with the running Datasette version and declaring the `semantic conventions schema <https://opentelemetry.io/docs/specs/otel/schemas/>`__ its attribute names follow.
This is separate from, and does not replace, the built-in :ref:`internals_tracer` mechanism behind ``?_trace=1`` and the :ref:`setting_trace_debug` setting. Both continue to work exactly as before.
.. _internals_telemetry_turning_on:
Turning tracing on
------------------
Install an OpenTelemetry SDK, an exporter and the instrumentation agent, then launch Datasette through ``opentelemetry-instrument``:
.. code-block:: bash
pip install opentelemetry-distro opentelemetry-exporter-otlp
OTEL_SERVICE_NAME=datasette \
OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 \
OTEL_METRICS_EXPORTER=none \
OTEL_LOGS_EXPORTER=none \
opentelemetry-instrument datasette mydb.db
Point ``OTEL_EXPORTER_OTLP_ENDPOINT`` at whichever tracing backend you use. To print spans straight to the terminal instead, with no backend at all, drop that variable and set ``OTEL_TRACES_EXPORTER=console`` in its place.
A few things catch people out the first time:
.. warning::
``OTEL_TRACES_EXPORTER=console datasette mydb.db`` produces **nothing**. That environment variable is read by the OpenTelemetry SDK's auto-configuration, which only runs when the ``opentelemetry-instrument`` agent wraps the process. Datasette core installs no provider, so a plain ``datasette`` process emits nothing at all, whatever ``OTEL_`` variables are set.
Spans do not appear immediately. The SDK's default ``BatchSpanProcessor`` flushes on a timer, every 5 seconds. Either wait, or stop the process - shutdown triggers a final flush - or set ``OTEL_BSP_SCHEDULE_DELAY=1000`` while you are experimenting. That last one is for demos, not for production.
Always set ``OTEL_SERVICE_NAME``. Without it the SDK's default resource reports a ``service.name`` of ``unknown_service``, and your traces will be filed under that instead of under a name you can search for.
Setting ``OTEL_METRICS_EXPORTER=none`` and ``OTEL_LOGS_EXPORTER=none`` is worth doing unless your backend accepts those signals too - ``opentelemetry-distro`` defaults every signal to OTLP, and a traces-only backend will reject the other two noisily. Datasette itself emits no metrics and no logs through OpenTelemetry.
Span reference
--------------
Datasette emits five spans. Four of them describe the database layer - one per query, one for the work that query does inside a SQL worker thread, and two more for the write queue - and the fifth covers startup. Attribute names use the ``datasette.*`` prefix for Datasette-specific data, alongside standard OpenTelemetry attributes such as ``db.system``.
This reference is generated from ``datasette/telemetry_registry.py``, the single source of truth for every span and attribute Datasette emits. A conformance test makes real requests and compares what is actually emitted against that registry in both directions, so nothing here is hand-maintained and nothing can silently drift out of date.
Spans are ``SpanKind.INTERNAL`` unless a kind is listed below. Only ``db.query`` is ``CLIENT``: it is the one span that represents a call to a database rather than Datasette's own work, and trace UIs use the kind to decide whether to render a span as a database call. Its children stay ``INTERNAL`` because they are Datasette's decomposition of that one query - marking them ``CLIENT`` too would make a single query look like several database calls to anything counting by kind.
.. [[[cog
from telemetry_doc import spans
spans(cog)
.. ]]]
``db.query``
A SQL operation issued by Datasette, covering the full round trip including any time spent queued for a thread.
Kind: ``CLIENT``.
Attributes:
- ``db.system`` - Always ``sqlite``.
- ``db.namespace`` - Name of the database being queried.
- ``db.query.text`` - The SQL, truncated to 2048 characters. Never the parameter values.
- ``db.operation.name`` *(optional)* - The statement's leading keyword - ``SELECT``, ``INSERT``, ``CREATE``, and so on - matched against a small fixed allowlist. Omitted rather than set to an arbitrary value: the allowlist exists because this attribute is a candidate dimension for a query-duration metric in a later phase, and echoing an unrecognised first token from user-supplied SQL would be an unbounded-cardinality hazard. Also omitted for ``execute_write_script()``, which runs multiple statements - per semantic conventions, the operation name should not be extracted from query text that can contain more than one operation. Note that a statement beginning with a CTE reports ``WITH``, not the operation inside it - a substantial share of Datasette's own reads take that form. Resolving it further would mean parsing.
- ``db.collection.name`` *(optional)* - The primary table, set only where the view already knows it - the table and row pages. Omitted for arbitrary ``?sql=`` queries, where determining the table would mean parsing the query.
- ``datasette.param_count`` *(optional)* - Number of bound parameters. Recorded instead of the values themselves.
- ``datasette.param_sets`` *(optional)* - Number of parameter sets consumed by ``execute_write_many()``. Not a row count - ``executemany()`` returns no rows. The parameter values themselves are never recorded: that sequence can hold thousands of rows.
- ``datasette.time_limit_ms`` *(optional)* - The :ref:`setting_sql_time_limit_ms` value this query ran under. Set on reads, which are the queries that time limit applies to.
- ``datasette.rows_returned`` *(optional)* - Number of rows a read returned. Set on the read path only, and only when the read succeeded.
- ``datasette.truncated`` *(optional)* - True if the result was cut short by :ref:`setting_max_returned_rows`.
- ``datasette.interrupted`` *(optional)* - True if the query was cancelled for exceeding the time limit. The span status is also set to ``ERROR``, unless the caller asked for a budget shorter than :ref:`setting_sql_time_limit_ms` - as table counts, facet suggestion and autocomplete all do - in which case running out of time is an expected answer rather than a failure and the status is left unset.
- ``datasette.sql_error_suppressed`` *(optional)* - True when the query failed but the caller passed ``log_sql_errors=False``, meaning it was probing and treats failure as an expected answer. Facet suggestion does this against every column.
- ``datasette.executescript`` *(optional)* - True for ``execute_write_script()``, which runs multiple statements.
- ``datasette.executemany`` *(optional)* - True for ``execute_write_many()``, which runs one statement against many parameter sets.
``db.query.execute``
The read executing inside a SQL worker thread. Child of ``db.query``; the gap between the two is time spent waiting for a thread.
No attributes.
``db.write.queue_wait``
Time a write spent waiting in its database's write queue before the write thread picked it up. Child of ``db.query`` for a ``block=True`` write, where the caller awaits the write and containment is accurate. For a ``block=False`` write the caller does not await it - the enqueueing request *caused* the write without *containing* it, and the write's spans can outlive the request's own - so this is a root span instead, carrying an OpenTelemetry link back to the enqueueing span rather than a parent. A link records causation without asserting containment, which is exactly the distinction here.
No attributes.
``db.write.execute``
The write executing on the write thread. Child of ``db.query`` for a ``block=True`` write; for ``block=False`` a root span with a link back to the enqueueing span instead - see ``db.write.queue_wait`` above.
Attributes:
- ``datasette.isolated_connection`` - True if the write ran on its own connection rather than the shared write connection.
- ``datasette.transaction`` - False for statements such as ``VACUUM`` that cannot run inside a transaction.
``datasette.startup``
``invoke_startup()`` running: ``register_events``, ``register_actions``, ``register_column_types``, ``prepare_jinja2_environment``, internal-database schema catalog refresh (including the ``prepare_connection`` warm-up this triggers for each database touched for the first time), saved queries, column type config and the ``startup`` hook. Runs once per process, before any request exists, so without this span every child it creates would be its own orphan root trace. A connection warmed later - lazily, the first time a *request* touches a new database or thread - nests under that request's own span instead, not under this one, since this span has already ended by then.
No attributes.
.. [[[end]]]
.. _internals_telemetry_privacy:
Privacy and safety
------------------
Spans leave your infrastructure whenever you configure an exporter, so what goes into them is a security decision. Datasette's rules are:
- **SQL text is truncated to 2048 characters.** On a public instance the SQL is supplied by visitors and is unbounded in length, so ``db.query.text`` is cut off - with a ``…[truncated]`` marker - rather than allowed to set the size of a span.
- **SQL parameter values are never recorded.** Only ``datasette.param_count``, a count. Parameter values are the part of a query most likely to hold something sensitive, and separating them from the SQL is the reason bound parameters exist.
- **No actor identifiers are recorded.** No actor ID, no actor JSON, no client IP address. Nothing on a span identifies who made the request.
- **Table names come only from an explicit** ``table=`` **argument.** ``db.collection.name`` is set by callers that already know which table they are working with, and is never derived from the SQL. Deriving it would mean parsing, and on an instance where visitors can create tables the set of possible values has no ceiling.
The SQL itself, though, *is* recorded, and on a public instance that means anything a visitor types into the query editor or passes as ``?sql=`` will be exported along with the span. That is the trade-off tracing a query engine makes.
.. _internals_telemetry_limitations:
Known limitations
-----------------
- **Datasette does not create a span for the HTTP request itself.** Every span listed above is therefore a root span unless something above Datasette - an ASGI instrumentation layer, or the web framework embedding it - has already started one for the request, in which case Datasette's spans nest underneath it correctly.
- **Two plugin hooks run outside the** ``datasette.startup`` **span.** ``register_output_renderer`` is dispatched from ``Datasette.__init__()`` and ``asgi_wrapper`` from ``Datasette.app()``, both of which happen before ``invoke_startup()``. Datasette itself queries no database in either, so a default install emits nothing there - but a plugin that does will produce a root trace. Covering these would mean holding a span open across object construction, which is worse than the orphan.
- ``db.operation.name`` **reports** ``WITH`` **for a statement that opens with a common table expression**, rather than the operation inside it, and a substantial share of Datasette's own reads take that form. The attribute is a leading-keyword match against a fixed allowlist, deliberately not a parse.
- **Spans emitted before a provider is installed are not recorded.** If you are embedding Datasette in a host application, install your ``TracerProvider`` before serving traffic. This is ordinary OpenTelemetry behaviour rather than anything Datasette controls; nothing is permanently affected, those particular spans are simply dropped.
.. _internals_csrf:
CSRF protection

37
docs/telemetry_doc.py Normal file
View file

@ -0,0 +1,37 @@
"""
Render the span reference in ``internals.rst`` from
``datasette/telemetry_registry.py``.
Driven by cog, and ``cog --check docs/*.rst`` runs in CI - so adding a span
without documenting it, or documenting one that no longer exists, is a build
failure rather than something a reader discovers later.
"""
def _attribute_lines(cog, attributes):
if not attributes:
cog.out(" No attributes.\n\n")
return
cog.out(" Attributes:\n\n")
for attribute in attributes:
suffix = " *(optional)*" if attribute.optional else ""
cog.out(f" - ``{attribute}``{suffix} - {attribute.description}\n")
cog.out("\n")
def spans(cog):
from opentelemetry.trace import SpanKind
from datasette.telemetry_registry import SPANS
cog.out("\n")
for span in SPANS:
title = f"{span}*" if span.prefix else str(span)
cog.out(f"``{title}``\n")
cog.out(f" {span.description}\n\n")
# INTERNAL is the default and the overwhelming majority of spans -
# printing it on every one would be noise. Only the exceptional case,
# a real database call, is worth calling out.
if span.kind != SpanKind.INTERNAL:
cog.out(f" Kind: ``{span.kind.name}``.\n\n")
_attribute_lines(cog, span.attributes)

2
pyproject.toml
View file

@ -40,6 +40,7 @@ dependencies = [
"setuptools",
"pip",
"pydantic>=2",
"opentelemetry-api>=1.37",
]
[project.urls]
@ -70,6 +71,7 @@ dev = [
"cogapp>=3.3.0",
"multipart-form-data-conformance==0.1a0",
"ruff>=0.16.0",
"opentelemetry-sdk>=1.37",
# docs
"Sphinx==7.4.7",
"furo==2025.9.25",

62
tests/conftest.py
View file

@ -58,6 +58,62 @@ def find_free_port():
return sock.getsockname()[1]
_otel_span_exporter = None
@pytest.fixture(scope="session", autouse=True)
def _otel_provider():
"""
Install a real OTel SDK TracerProvider + InMemorySpanExporter exactly
once, before any span is ever created in this process.
This has to be session-scoped and autouse because
`opentelemetry.trace.set_tracer_provider()` is effectively
once-per-process: a second call logs a warning and is ignored. So the
install must happen exactly once, before anything asserts on spans.
`datasette.telemetry.tracer` is a module-level `ProxyTracer`. Once a
provider exists, the first span it starts resolves a concrete tracer
and caches it permanently. It does *not* cache the no-op tracer, so
any span started before this fixture runs is merely lost rather than
poisoning the tracer for the rest of the process. If the SDK isn't
installed, do nothing: core spans stay no-op `NonRecordingSpan`s and
the rest of the suite is unaffected.
"""
global _otel_span_exporter
try:
from opentelemetry import trace as otel_trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
InMemorySpanExporter,
)
except ImportError:
return
exporter = InMemorySpanExporter()
provider = TracerProvider()
# SimpleSpanProcessor exports synchronously on span end - no background
# batching thread, so assertions immediately after a request never race.
provider.add_span_processor(SimpleSpanProcessor(exporter))
otel_trace.set_tracer_provider(provider)
_otel_span_exporter = exporter
@pytest.fixture
def otel_spans():
"""
Function-scoped access to the finished-spans exporter: clears any spans
left over from previous tests, then yields the exporter so a test can
call `.get_finished_spans()` after making requests. Skips (rather than
fails) if the OTel SDK is not installed.
"""
pytest.importorskip("opentelemetry.sdk")
if _otel_span_exporter is None:
pytest.skip("OpenTelemetry SDK provider was not installed")
_otel_span_exporter.clear()
yield _otel_span_exporter
@pytest.fixture
def bare_ds():
"""
@ -168,6 +224,12 @@ def pytest_collection_modifyitems(config, items):
move_to_front(items, "test_spatialite_error_if_attempt_to_open_spatialite")
move_to_front(items, "test_package")
move_to_front(items, "test_package_with_port")
# Same reason: this one shells out to a fresh interpreter. Late in a serial
# run the pytest process holds enough threads that the fork half of
# subprocess' fork+exec crashes the interpreter on macOS/CPython 3.13
# (SIGSEGV/SIGBUS inside _execute_child). Reproduces with any subprocess
# call placed there, on an unmodified tree - running it first avoids it.
move_to_front(items, "test_datasette_package_never_imports_the_sdk")
def move_to_front(items, test_name):

109
tests/test_internals_database.py
View file

@ -3,11 +3,13 @@ Tests for the datasette.database.Database class
"""
import asyncio
import threading
import uuid
from types import SimpleNamespace
import pytest
import sqlite_utils
from opentelemetry import context as otel_context_api
from datasette.app import Datasette
from datasette.database import (
@ -1223,3 +1225,110 @@ async def test_database_close_is_idempotent(tmpdir):
# Second call should be a no-op, not raise
db.close()
ds._internal_database.close()
_CONTEXT_LEAK_MARKER_KEY = "otel-context-leak-marker"
@pytest.mark.asyncio
@pytest.mark.parametrize("num_sql_threads", (0, 1))
async def test_write_thread_context_is_detached_between_tasks(
tmp_path, monkeypatch, num_sql_threads
):
"""
The write thread attaches each task's otel Context and must detach it
again before picking up the next task. The thread is persistent and
shared, so a leaked token would grow that thread's context stack for the
rest of the process - and a *wrong*-token detach only logs a warning
rather than raising, so "does it throw" cannot catch either mistake.
Two things are asserted, because neither alone is sufficient:
1. Each task observes the context value that was current on the event
loop when it was queued. This is what fails if the Context is not
carried on WriteTask, or is never attached. It does *not* catch a
missing detach: attach() replaces the current Context wholesale, so a
leftover one from a previous task is simply overwritten.
2. The write thread's attach depth is identical at the same point in
every task. This is what fails if detach is missing - the stack grows
by one per task - and it holds across a task that raises, because the
detach lives in a `finally`.
An otel context value is used rather than a plain contextvars.ContextVar:
a plain var set on the event loop never crosses into the write thread, so
the probe would read None every time and the test could not fail.
"""
name = f"context_leak_test_{num_sql_threads}"
db_path = tmp_path / f"{name}.db"
sqlite3.connect(db_path).close()
ds = Datasette([str(db_path)], settings={"num_sql_threads": num_sql_threads})
db = ds.get_database(name)
await db.execute_write("create table t (id integer primary key)")
write_thread_name = f"_execute_writes for database {name}"
depth = {"value": 0}
real_attach = otel_context_api.attach
real_detach = otel_context_api.detach
def counting_attach(context):
token = real_attach(context)
if threading.current_thread().name == write_thread_name:
depth["value"] += 1
return token
def counting_detach(token):
real_detach(token)
if threading.current_thread().name == write_thread_name:
depth["value"] -= 1
# Patched on the opentelemetry.context module itself, which is what both
# database.py and opentelemetry.trace.use_span() look the functions up on.
monkeypatch.setattr(otel_context_api, "attach", counting_attach)
monkeypatch.setattr(otel_context_api, "detach", counting_detach)
seen_markers = []
seen_depths = []
def probe(conn):
seen_markers.append(otel_context_api.get_value(_CONTEXT_LEAK_MARKER_KEY))
seen_depths.append(depth["value"])
def failing_probe(conn):
probe(conn)
# Exercises the write thread's exception path: the detach still has
# to happen, which is why it lives in a `finally`.
raise ValueError("deliberate failure inside a write task")
try:
for i in range(5):
ctx = otel_context_api.set_value(_CONTEXT_LEAK_MARKER_KEY, f"marker-{i}")
token = real_attach(ctx)
try:
if i == 2:
with pytest.raises(ValueError):
await db.execute_write_fn(failing_probe)
else:
await db.execute_write_fn(probe)
finally:
real_detach(token)
# Sanity check: no marker is active in *this* (event loop) context
# right now, so the final probe is a fair test of the write thread's
# own state rather than something this test forgot to clean up.
assert otel_context_api.get_value(_CONTEXT_LEAK_MARKER_KEY) is None
await db.execute_write_fn(probe)
finally:
db.close()
assert seen_markers == [
"marker-0",
"marker-1",
"marker-2",
"marker-3",
"marker-4",
None,
]
assert len(set(seen_depths)) == 1, (
f"write thread context stack grew across tasks: {seen_depths} - "
"a token was attached without being detached"
)

1015
tests/test_telemetry.py Normal file
View file

File diff suppressed because it is too large Load diff

309
tests/test_telemetry_registry.py Normal file
View file

@ -0,0 +1,309 @@
"""
Two-way conformance between `datasette/telemetry_registry.py` and what
Datasette actually emits.
This is the test that makes the generated documentation trustworthy. cog
guarantees the docs match the registry; this guarantees the registry matches
the code. Without it, both could agree with each other and be wrong.
It checks both directions, and the second one is the one nothing else catches:
- **emitted but not registered** - instrumentation was added without
documenting it, so the reference page silently omits it.
- **registered but never emitted** - the reference page describes a span or
attribute that no longer exists, which is worse than omitting it, because a
reader will build a dashboard on it.
Both of those directions compare the code against the registry. Neither can
catch a *rename*, because the call sites now take their names from the
registry - move `DB_NAMESPACE` to `"db.namespace2"` and code and registry
still agree with each other, while every existing dashboard breaks. So the
literal names live here too, spelled out, and are asserted against both the
registry and the wire. That is the one comparison in this file that is not
made against a value derived from the registry itself.
"""
import itertools
import pytest
import pytest_asyncio
pytest.importorskip("opentelemetry.sdk")
from datasette import telemetry_registry as reg
from datasette.app import Datasette
from datasette.database import QueryInterrupted
from datasette.utils.sqlite import sqlite3
# The names as they appear on the wire, written out rather than read from the
# registry. If a change to the registry makes one of these fail, that change
# is renaming something a user's dashboards and saved queries depend on -
# which is a decision to take deliberately, here, not a line to re-derive.
EXPECTED_ATTRIBUTES = {
"db.query": {
"db.system",
"db.namespace",
"db.query.text",
"db.operation.name",
"db.collection.name",
"datasette.param_count",
"datasette.param_sets",
"datasette.time_limit_ms",
"datasette.rows_returned",
"datasette.truncated",
"datasette.interrupted",
"datasette.sql_error_suppressed",
"datasette.executescript",
"datasette.executemany",
},
"db.query.execute": set(),
"db.write.queue_wait": set(),
"db.write.execute": {
"datasette.isolated_connection",
"datasette.transaction",
},
"datasette.startup": set(),
}
EXPECTED_SPANS = set(EXPECTED_ATTRIBUTES)
# Named in-memory databases are shared-cache, so two Datasette instances using
# the same name share one SQLite database - and the second `create table`
# fails. Every workload below therefore gets its own name.
_names = itertools.count()
def _unique(prefix):
return f"{prefix}{next(_names)}"
async def exercise():
"""
Drive enough of Datasette to emit every span and attribute the registry
claims exists.
Each call is here because it is the only thing that produces some span or
attribute - see the comments. If you add instrumentation on a path this
does not reach, add the path rather than loosening the assertions.
Returns the instance so the caller can close it; startup happens inside
so that the `datasette.startup` span lands in the collected set.
"""
name = _unique("registry")
ds = Datasette(memory=True)
ds.add_memory_database(name)
# datasette.startup - and the internal catalog work nested under it
await ds.invoke_startup()
db = ds.get_database(name)
# Writes: db.write.queue_wait, db.write.execute, db.query
await db.execute_write("create table t (id integer primary key, v text)")
# datasette.executemany, datasette.param_sets
await db.execute_write_many(
"insert into t (id, v) values (?, ?)", [[i, f"v{i}"] for i in range(30)]
)
# datasette.executescript
await db.execute_write_script("create table t2 (id integer); drop table t2;")
# datasette.transaction=False - VACUUM cannot run inside a transaction
await db.execute_write("vacuum", transaction=False)
# datasette.isolated_connection=True
await db.execute_isolated_fn(lambda conn: conn.execute("select 1").fetchone())
# Reads: db.query.execute, datasette.rows_returned, datasette.truncated,
# datasette.param_count, datasette.time_limit_ms
await db.execute("select * from t where id > :n", {"n": 5})
await db.execute("select * from t", truncate=True)
# datasette.sql_error_suppressed - the caller is probing and treats
# failure as an expected answer
with pytest.raises(sqlite3.OperationalError):
await db.execute("select nope from t", log_sql_errors=False)
# datasette.interrupted - only ever set when a query exceeds its time
# limit, so the workload has to force one rather than exempt it. An
# unbounded recursive CTE cannot finish, so 1ms is always exceeded.
with pytest.raises(QueryInterrupted):
await db.execute(
"with recursive c(x) as (select 0 union all select x+1 from c) "
"select * from c",
custom_time_limit=1,
)
# db.collection.name - set only by views that already know their table
assert (await ds.client.get(f"/{name}/t?_facet=v")).status_code == 200
assert (await ds.client.get(f"/{name}/t/1.json")).status_code == 200
return ds
@pytest_asyncio.fixture
async def emitted(otel_spans):
"Every span name and (span name, attribute key) pair a broad workload emits."
# otel_spans has already cleared the exporter, and nothing is cleared
# after this point: the workload's own startup emits datasette.startup.
ds = await exercise()
spans = otel_spans.get_finished_spans()
assert spans, "no spans captured - the fixture is not exercising anything"
names = set()
pairs = set()
for span in spans:
# str() because span.name is the registry's SpanName instance, and a
# set of those would compare equal to literals but read confusingly
# in a failure message.
names.add(str(span.name))
for key in span.attributes or {}:
pairs.add((str(span.name), str(key)))
ds.close()
return {"names": names, "pairs": pairs}
def _keys_by_span(pairs):
by_span = {}
for span_name, key in pairs:
by_span.setdefault(span_name, set()).add(key)
return by_span
@pytest.mark.asyncio
async def test_workload_emits_exactly_the_expected_names(emitted):
"""
The wire format, pinned to literals.
Not derived from the registry, so this is what catches a rename that the
registry and the call sites make together.
"""
assert emitted["names"] == EXPECTED_SPANS
by_span = _keys_by_span(emitted["pairs"])
assert {name: by_span.get(name, set()) for name in emitted["names"]} == (
EXPECTED_ATTRIBUTES
)
def test_registry_matches_the_expected_names():
"The other half of the rename check: the registry against the same literals."
assert {str(span) for span in reg.SPANS} == EXPECTED_SPANS
for span in reg.SPANS:
assert {str(attribute) for attribute in span.attributes} == EXPECTED_ATTRIBUTES[
str(span)
], f"{span} attributes have drifted"
@pytest.mark.asyncio
async def test_every_emitted_span_is_registered(emitted):
"A span added without a registry entry would be missing from the docs."
unregistered = sorted(
name for name in emitted["names"] if reg.span_for(name) is None
)
assert (
not unregistered
), f"these spans are emitted but not in telemetry_registry.SPANS: {unregistered}"
@pytest.mark.asyncio
async def test_every_emitted_attribute_is_registered(emitted):
"An attribute added without a registry entry would be missing from the docs."
unregistered = sorted(
f"{span_name} -> {key}"
for span_name, key in emitted["pairs"]
if not reg.attribute_allowed(reg.span_for(span_name), key)
)
assert (
not unregistered
), "these span attributes are emitted but not registered: " + ", ".join(
unregistered
)
@pytest.mark.asyncio
async def test_every_registered_span_is_emitted(emitted):
"""
The direction nothing else catches: the docs must not describe a span that
no longer exists.
"""
missing = sorted(
str(span)
for span in reg.SPANS
if not any(reg.span_for(name) is span for name in emitted["names"])
)
assert not missing, (
f"these spans are documented but never emitted by the workload: {missing}. "
"Either the instrumentation was removed, or exercise() no longer reaches it."
)
@pytest.mark.asyncio
async def test_every_registered_attribute_is_emitted(emitted):
"""
Every registered attribute, optional or not, must actually be set at least
once by the workload.
`optional` describes whether a reader should expect it on every span, not
whether the code still sets it - so an attribute deleted from the code but
left in the docs has to fail here even when it is marked optional. If a
new attribute only appears in some rare case, extend exercise() to reach
that case.
"""
by_span = _keys_by_span(emitted["pairs"])
missing = []
for span in reg.SPANS:
emitted_keys = by_span.get(str(span), set())
for attribute in span.attributes:
if attribute not in emitted_keys:
missing.append(f"{span} -> {attribute}")
assert not missing, (
"these attributes are documented but never emitted by the workload: "
+ ", ".join(sorted(missing))
)
def test_registry_has_no_duplicate_names():
assert len(set(reg.SPANS)) == len(reg.SPANS)
for span in reg.SPANS:
assert len(set(span.attributes)) == len(
span.attributes
), f"{span} lists an attribute twice"
def test_registry_entries_are_documented():
"Every entry carries a description - the docs are generated from these."
for span in reg.SPANS:
assert span.description.strip(), f"{span} has no description"
for attribute in span.attributes:
assert attribute.description.strip(), f"{span} -> {attribute} has none"
def test_registry_entries_are_usable_as_plain_strings():
"The str subclassing is the whole reason call sites need no wrapper API."
assert isinstance(reg.DB_QUERY, str)
assert isinstance(reg.DB_NAMESPACE, str)
assert reg.DB_QUERY == "db.query"
assert reg.DB_NAMESPACE == "db.namespace"
assert f"{reg.DB_QUERY}.execute" == "db.query.execute"
def test_span_and_attribute_lookup():
assert reg.span_for("db.query") is reg.DB_QUERY
assert reg.span_for("datasette.startup") is reg.STARTUP
assert reg.span_for("not.a.datasette.span") is None
assert reg.attribute_allowed(reg.DB_QUERY, "db.namespace")
assert not reg.attribute_allowed(reg.DB_QUERY, "db.namespace.extra")
assert not reg.attribute_allowed(reg.DB_QUERY, "datasette.isolated_connection")
assert not reg.attribute_allowed(None, "db.namespace")
def test_prefix_span_lookup():
"""
`prefix=True` matching, exercised directly.
Phase 1 registers no prefix spans, so without this the branch in
`span_for()` would be untested code that the conformance tests silently
never reach.
"""
hook = reg.SpanName("datasette.hook.", "A hypothetical span family", prefix=True)
original = reg.SPANS
reg.SPANS = original + (hook,)
try:
assert reg.span_for("datasette.hook.render_cell") is hook
assert reg.span_for("datasette.hook.anything") is hook
assert reg.span_for("datasette.hookish") is None
assert reg.span_for("db.query") is reg.DB_QUERY
finally:
reg.SPANS = original