remote/datasette
1
0
Fork 0
mirror of https://github.com/simonw/datasette.git synced 2026-10-18 23:36:57 +02:00
Code Issues Projects Releases Packages Wiki Activity

Connection budget: pages, startup and sweeps only touch the databases they need

Browse source 
Adds tests/connection_budget.py, a helper that records which databases a
block of code opens, reads, writes or ATTACHes, and tests/test_connection_budget.py
which asserts budgets for ~60 endpoints, startup, sweeps and telemetry with
50 attached databases.

Fixes for every all-database loop it found:

- Index page: tables, views, hidden tables, columns, primary keys, FTS
  tables and foreign-key relationship counts come from the _internal
  catalog (a few queries for any number of databases). Row counts are only
  run when at most 20 databases need counting (COUNT_MAX_DATABASES);
  above that the page shows table counts plus counts already known.
- Derived-table permissions: allowed_resources() no longer queries every
  database per page; dependency maps are computed by the SchemaWatcher from
  the rows it scans, cached per database, and read back from the catalog
  for databases restored from a persisted catalog.
- catalog_tables gains a type column (PRAGMA table_list: table, virtual,
  shadow) so hidden shadow tables can be identified without a connection.
- PaginatedResources.all() fetches pages of 1000 after the first.
- /-/api and /-/debug/autocomplete use the catalog; /-/schema does one
  permission query and skips databases with nothing visible.
- /-/versions computes SQLite details once per instance, without crossdb
  ATTACHes; detect_json1() caches its answer.
- /-/databases only stats/hashes databases the actor can see.
- Immutable table counts are computed when first needed, not at startup.
- check_databases() (serve, --get) reuses the catalog scan instead of
  opening every database a second time, and reports "file is not a
  database" as a usage error.
- SchemaWatcher: failed scans are retried when the file changes, not on
  every sweep; scan connections skip crossdb ATTACH; catalog_generation
  counter for caches.
- The SQL write UI no longer forces a stat sweep of every database.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EwkLo2rJWidpkDVT3HkvNv
This commit is contained in:
Claude 2026-10-04 05:58:31 +00:00
commit 1bc15caef7
No known key found for this signature in database
17 changed files with 1768 additions and 232 deletions
Download patch file Download diff file Expand all files Collapse all files

150
datasette/app.py
View file

@ -8,6 +8,7 @@ if TYPE_CHECKING:
from datasette.permissions import Resource
from datasette.tokens import TokenRestrictions
import collections
import copy
import dataclasses
import datetime
import functools
@ -109,6 +110,7 @@ from .utils.asgi import (
asgi_send_redirect,
asgi_static,
)
from .utils.catalog import all_derived_table_dependencies
from .utils.internal_db import init_internal_db
from .utils.sqlite import (
sqlite3,
@ -507,6 +509,11 @@ class Datasette:
self.nolock = nolock
self.internal_db_created = False
self._schema_watcher = SchemaWatcher(self)
# (catalog generation, derived-table dependency map) - see
# datasette.utils.catalog.all_derived_table_dependencies()
self._catalog_derived_cache = None
# SQLite version/extension details for /-/versions, computed once
self._sqlite_versions_info = None
if memory or crossdb or not self.files:
self.add_database(
Database(self, is_mutable=False, is_memory=True), name="_memory"
@ -1616,7 +1623,7 @@ class Datasette:
separator = "&" if "?" in url else "?"
return url + separator + urllib.parse.urlencode({"_hash": hash_value})
def _prepare_connection(self, conn, database):
def _prepare_connection(self, conn, database, *, crossdb=True):
conn.row_factory = sqlite3.Row
conn.text_factory = lambda x: str(x, "utf-8", "replace")
if self.sqlite_extensions and database != INTERNAL_DB_NAME:
@ -1648,7 +1655,7 @@ class Datasette:
if database != INTERNAL_DB_NAME:
pm.hook.prepare_connection(conn=conn, database=database, datasette=self)
# If self.crossdb and this is _memory, connect the first SQLITE_LIMIT_ATTACHED databases
if self.crossdb and database == "_memory":
if crossdb and self.crossdb and database == "_memory":
count = 0
for db_name, db in self.databases.items():
if count >= SQLITE_LIMIT_ATTACHED or db.is_memory:
@ -1873,19 +1880,34 @@ class Datasette:
parent,
include_is_private,
):
databases = (
[(parent, self.databases[parent])]
if parent in self.databases
else ([] if parent is not None else list(self.databases.items()))
)
dependency_maps = dict(
zip(
(name for name, _ in databases),
await asyncio.gather(
*(db.derived_table_dependencies() for _, db in databases)
),
if parent is not None:
dependency_maps = (
{parent: await self.databases[parent].derived_table_dependencies()}
if parent in self.databases
else {}
)
)
else:
# Every database: read the dependencies from the catalog (cached
# until it changes) rather than asking each database in turn -
# that was a query against every attached database for every
# page of allowed_resources(). The listing being filtered comes
# from the same catalog rows. Only databases the SchemaWatcher
# does not track fall back to live introspection.
dependency_maps = dict(await all_derived_table_dependencies(self))
unwatched = [
(name, db)
for name, db in self.databases.items()
if db._watch_state is None
]
if unwatched:
dependency_maps.update(
zip(
(name for name, _ in unwatched),
await asyncio.gather(
*(db.derived_table_dependencies() for _, db in unwatched)
),
)
)
dependencies = [
(database_name, child, source)
for database_name, dependency_map in dependency_maps.items()
@ -2420,7 +2442,7 @@ ORDER BY allowed.parent, allowed.child
url = "https://" + url[len("http://") :]
return url
def _connected_databases(self):
def _connected_databases(self, names=None):
return [
{
"name": d.name,
@ -2432,23 +2454,58 @@ ORDER BY allowed.parent, allowed.child
"hash": d.hash,
}
for name, d in self.databases.items()
if names is None or name in names
]
async def _connected_databases_for_actor(self, actor):
page = await self.allowed_resources("view-database", actor)
allowed_names = {resource.parent async for resource in page.all()}
return [
database
for database in self._connected_databases()
if database["name"] in allowed_names
]
# Only stat (size) and hash (a full read of an immutable file, the
# first time) the databases this actor may see
return self._connected_databases(allowed_names)
async def _databases_data(self, request):
return {"databases": await self._connected_databases_for_actor(request.actor)}
def _versions(self):
if self._sqlite_versions_info is None:
self._sqlite_versions_info = self._sqlite_versions()
sqlite_info, pysqlite3_version = self._sqlite_versions_info
datasette_version = {"version": __version__}
if self.version_note:
datasette_version["note"] = self.version_note
try:
# Optional import to avoid breaking Pyodide
# https://github.com/simonw/datasette/issues/1733#issuecomment-1115268245
import uvicorn
uvicorn_version = uvicorn.__version__
except ImportError:
uvicorn_version = None
info = {
"python": {
"version": ".".join(map(str, sys.version_info[:3])),
"full": sys.version,
},
"datasette": datasette_version,
"asgi": "3.0",
"uvicorn": uvicorn_version,
"sqlite": copy.deepcopy(sqlite_info),
}
if pysqlite3_version is not None:
info["pysqlite3"] = pysqlite3_version
return info
def _sqlite_versions(self):
"""SQLite version, extensions, FTS versions and compile options, as
seen by a fully prepared connection. They cannot change while the
process runs, so /-/versions computes this once per instance rather
than preparing (loading extensions into) a connection per request.
crossdb=False: ATTACHing every --crossdb database is not needed to
report versions."""
conn = sqlite3.connect(":memory:")
self._prepare_connection(conn, "_memory")
self._prepare_connection(conn, "_memory", crossdb=False)
sqlite_version = conn.execute("select sqlite_version()").fetchone()[0]
sqlite_extensions = {"json1": detect_json1(conn)}
for extension, testsql, hasversion in (
@ -2482,44 +2539,24 @@ ORDER BY allowed.parent, allowed.child
fts_versions.append(fts)
except sqlite3.OperationalError:
continue
datasette_version = {"version": __version__}
if self.version_note:
datasette_version["note"] = self.version_note
try:
# Optional import to avoid breaking Pyodide
# https://github.com/simonw/datasette/issues/1733#issuecomment-1115268245
import uvicorn
uvicorn_version = uvicorn.__version__
except ImportError:
uvicorn_version = None
info = {
"python": {
"version": ".".join(map(str, sys.version_info[:3])),
"full": sys.version,
},
"datasette": datasette_version,
"asgi": "3.0",
"uvicorn": uvicorn_version,
"sqlite": {
"version": sqlite_version,
"fts_versions": fts_versions,
"extensions": sqlite_extensions,
"compile_options": [
r[0] for r in conn.execute("pragma compile_options;").fetchall()
],
},
sqlite_info = {
"version": sqlite_version,
"fts_versions": fts_versions,
"extensions": sqlite_extensions,
"compile_options": [
r[0] for r in conn.execute("pragma compile_options;").fetchall()
],
}
conn.close()
pysqlite3_version = None
if using_pysqlite3:
for package in ("pysqlite3", "pysqlite3-binary"):
try:
info["pysqlite3"] = importlib.metadata.version(package)
pysqlite3_version = importlib.metadata.version(package)
break
except importlib.metadata.PackageNotFoundError:
pass
conn.close()
return info
return sqlite_info, pysqlite3_version
def _plugins(self, request=None, all=False):
ps = list(get_plugins())
@ -3135,11 +3172,10 @@ ORDER BY allowed.parent, allowed.child
if self._startup_invoked and self._setup_db_done:
return
if not self._setup_db_done:
# First time server starts up, calculate table counts for
# immutable databases
for database in self.databases.values():
if not database.is_mutable:
await database.table_counts(limit=60 * 60 * 1000)
# Immutable databases used to have their table counts
# computed here, opening every one of them at startup. They
# are now computed the first time a page needs them (see
# Database.table_counts()) and cached from then on.
self._setup_db_done = True
await self.invoke_startup()

52
datasette/cli.py
View file

@ -896,9 +896,22 @@ pm.hook.register_commands(cli=cli)
async def check_databases(ds):
# Run check_connection against every connected database
# to confirm they are all usable
for database in list(ds.databases.values()):
"""Confirm every connected database is usable before serving.
Building the _internal catalog already opens every database with a
prepared connection and reads every table's columns, which is what
check_connection() checks. So build it first, then re-check only:
* databases whose catalog scan failed, to turn the failure into a
helpful error (SpatiaLite not loaded, not a database...)
* databases restored from a persisted catalog without being opened that
contain virtual tables, whose modules may no longer be loaded
instead of opening every database a second time.
"""
await ds._refresh_schemas()
to_check = await _databases_needing_check(ds)
for database in to_check:
try:
await database.execute_fn(check_connection)
except SpatialiteConnectionProblem:
@ -926,3 +939,36 @@ async def check_databases(ds):
):
msg = f"Warning: --crossdb only works with the first {SQLITE_LIMIT_ATTACHED} attached databases"
click.echo(click.style(msg, bold=True, fg="yellow"), err=True)
async def _databases_needing_check(ds):
restored_with_virtual_tables = set()
restored = [
name
for name, db in ds.databases.items()
if db._watch_state is not None
and db._watch_state.error is None
and db._watch_state.stats["scans"] == 0
and db._watch_state.catalog_version is not None
]
if restored:
rows = await ds.get_internal_database().execute(
"""
select distinct database_name from catalog_tables
where coalesce(rootpage, 0) = 0
and database_name in (select value from json_each(:names))
""",
{"names": json.dumps(restored)},
)
restored_with_virtual_tables = {row[0] for row in rows.rows}
to_check = []
for name, db in list(ds.databases.items()):
state = db._watch_state
if (
state is None
or state.error is not None
or name in restored_with_virtual_tables
or (state.catalog_version is None and not state.missing)
):
to_check.append(db)
return to_check

49
datasette/database.py
View file

@ -65,6 +65,7 @@ from .utils import (
table_column_details,
table_columns,
)
from .utils.catalog import SPATIALITE_HIDDEN_TABLES
from .utils.sql_analysis import SQLAnalysis, analyze_sql_tables
from .utils.sqlite import sqlite_derived_table_dependencies, sqlite_hidden_table_names
@ -83,6 +84,11 @@ _SHUTDOWN = object()
class Database:
# For table counts stop at this many rows:
count_limit = 10000
# Counts for immutable databases are computed once, the first time a
# page needs them, with this per-table time limit, then cached (they
# cannot change). They used to be computed for every immutable database
# at startup.
immutable_count_time_limit_ms = 60 * 60 * 1000
def __init__(
self,
@ -1066,8 +1072,12 @@ class Database:
return self.cached_size
async def table_counts(self, limit=10):
if not self.is_mutable and self.cached_table_counts is not None:
return self.cached_table_counts
if not self.is_mutable:
if self.cached_table_counts is not None:
return self.cached_table_counts
# The result is cached for the life of the process, so do not
# let a caller's short limit cache timeouts (None) forever
limit = max(limit, self.immutable_count_time_limit_ms)
# Try to get counts for each table, $limit timeout for each count
counts = {}
for table in await self.table_names():
@ -1218,20 +1228,7 @@ class Database:
has_spatialite = await self.execute_fn(detect_spatialite)
if has_spatialite:
# Also hide Spatialite internal tables
hidden_tables += [
"ElementaryGeometries",
"SpatialIndex",
"geometry_columns",
"spatial_ref_sys",
"spatialite_history",
"sql_statements_log",
"sqlite_sequence",
"views_geometry_columns",
"virts_geometry_columns",
"data_licenses",
"KNN",
"KNN2",
] + [
hidden_tables += list(SPATIALITE_HIDDEN_TABLES) + [
r[0] for r in (await self.execute("""
select name from sqlite_master
where name like "idx_%"
@ -1250,15 +1247,17 @@ class Database:
and not state.needs_scan
and not state.missing
):
# O(1): the schema_version the SchemaWatcher last stored in the
# catalog. allowed_resources() calls this for every database on
# every page, so a PRAGMA here was a per-request sweep of all
# databases (one pooled connection each). The cache is now as
# fresh as the catalog itself. The scan count is part of the key
# because a replaced file can have the same schema_version.
schema_version = ("catalog", state.catalog_version, state.stats["scans"])
else:
schema_version = (await self.execute("PRAGMA schema_version")).first()[0]
# From the catalog rows (computed by the SchemaWatcher when it
# scanned this database, or read back from _internal), so a
# permission check on one of this database's tables does not
# open it and agrees with what allowed_resources() lists. As
# fresh as the catalog itself.
from .utils.catalog import catalog_derived_table_dependencies
return (await catalog_derived_table_dependencies(self.ds, [self]))[
self.name
]
schema_version = (await self.execute("PRAGMA schema_version")).first()[0]
if (
self._cached_derived_table_dependencies is None
or self._cached_derived_table_dependencies[0] != schema_version

61
datasette/schema_watcher.py
View file

@ -66,6 +66,7 @@ import os
import time
from .utils import sqlite3
from .utils.catalog import remember_derived_table_dependencies
from .utils.internal_db import CATALOG_TABLES, collect_schema, write_catalog_entries
logger = logging.getLogger("datasette.schema_watcher")
@ -241,6 +242,9 @@ class SchemaWatcher:
self._task = None
self._spawned = set()
self._last_sweep = 0.0
# Bumped after every write to the catalog tables, so readers can
# cache things derived from the catalog (see datasette.utils.catalog)
self.catalog_generation = 0
self.counters = {
"sweeps": 0,
"sweep_seconds": 0.0,
@ -519,7 +523,9 @@ class SchemaWatcher:
outcomes = []
file_candidates = [s for s in candidates if s.is_file]
if file_candidates:
outcomes = await self._off_loop(self._sweep_sync, file_candidates)
outcomes = await self._off_loop(
self._sweep_sync, file_candidates, background
)
for state in candidates:
if not state.is_file and state.db.memory_name and not state.removed:
outcomes.append(await self._memory_pragma_check(state))
@ -536,6 +542,8 @@ class SchemaWatcher:
state.db._file_seen = True
if kind == "clean":
pass
elif kind == "failed_unchanged":
continue
elif kind == "missing":
self.counters["missing"] += 1
if not state.missing:
@ -569,7 +577,7 @@ class SchemaWatcher:
self.counters["sweep_seconds"] += time.perf_counter() - t0
self._last_sweep = time.monotonic()
def _sweep_sync(self, states):
def _sweep_sync(self, states, background=False):
"""Runs in a worker thread. Reads state, never writes it."""
by_dir = {}
for state in states:
@ -593,6 +601,20 @@ class SchemaWatcher:
if fp == prev and not state.fp_racy and not state.needs_scan:
out.append((state, "clean", fp, t_ns, None))
continue
if (
background
and state.error is not None
and fp == prev
# Not state.fp_racy: failed checks set that flag to
# force a recheck, which would retry forever
and not is_racy(fp, t_ns)
):
# The last scan failed (missing extension module, not
# a database, ...) and the file has not changed since:
# retrying would fail again. An explicit
# refresh_schemas() still retries.
out.append((state, "failed_unchanged", fp, t_ns, None))
continue
if fp[0] is None:
out.append((state, "missing", fp, t_ns, None))
continue
@ -652,7 +674,9 @@ class SchemaWatcher:
conn = db.connect(track=False)
try:
if prepare:
self.ds._prepare_connection(conn, db.name)
# crossdb=False: a scan of _memory reads its own schema only,
# it must not ATTACH (open) up to ten other database files
self.ds._prepare_connection(conn, db.name, crossdb=False)
else:
conn.row_factory = None
except Exception:
@ -673,17 +697,22 @@ class SchemaWatcher:
fp = fingerprint(db.path)
if fp[0] is None:
return {"state": state, "missing": True, "fp": fp, "t": t_ns}
conn = self._connect(db, prepare=True)
try:
conn.row_factory = sqlite3.Row
conn.execute("BEGIN")
conn = self._connect(db, prepare=True)
try:
version = conn.execute("PRAGMA schema_version").fetchone()[0]
schema = collect_schema(conn, db.name)
conn.row_factory = sqlite3.Row
conn.execute("BEGIN")
try:
version = conn.execute("PRAGMA schema_version").fetchone()[0]
schema = collect_schema(conn, db.name)
finally:
conn.rollback()
finally:
conn.rollback()
finally:
self._close(db, conn)
self._close(db, conn)
except Exception as e: # noqa: BLE001
# Keep the fingerprint: a background sweep retries a failed scan
# only once the file changes, instead of every interval
return {"state": state, "error": e, "fp": fp, "t": t_ns}
return {
"state": state,
"fp": fp,
@ -806,6 +835,9 @@ class SchemaWatcher:
if "error" in r:
state.error = r["error"]
state.needs_scan = True
if r.get("fp") is not None and r["fp"][0] is not None:
state.fp = r["fp"]
state.fp_racy = is_racy(r["fp"], r["t"])
logger.warning(
"Could not read schema of database %r: %s", state.name, r["error"]
)
@ -838,6 +870,7 @@ class SchemaWatcher:
write_catalog_entries(conn, live)
await self.ds.get_internal_database().execute_write_fn(_write)
self.catalog_generation += 1
for r in stored:
state = r["state"]
state.catalog_version = r["version"]
@ -849,6 +882,9 @@ class SchemaWatcher:
state.db._file_seen = True
state.stats["scans"] += 1
self.counters["scans"] += 1
remember_derived_table_dependencies(
state, [(t[1], t[3]) for t in r["schema"]["tables"]]
)
if missing:
for state in missing:
if not state.missing:
@ -870,6 +906,7 @@ class SchemaWatcher:
)
await self.ds.get_internal_database().execute_write_fn(_clear)
self.catalog_generation += 1
for state in states:
state.catalog_version = None
@ -885,6 +922,7 @@ class SchemaWatcher:
conn.execute(f"DELETE FROM {table} WHERE database_name = ?", [name])
await self.ds.get_internal_database().execute_write_fn(_delete)
self.catalog_generation += 1
for name in names:
self._pending_removals.discard(name)
@ -912,6 +950,7 @@ class SchemaWatcher:
)
await internal.execute_write_fn(_delete_stale)
self.catalog_generation += 1
self._pending_removals.clear()
states = [s for s in self.states.values() if not s.removed]
persisted = {}

52
datasette/utils/__init__.py
View file

@ -34,6 +34,10 @@ if typing.TYPE_CHECKING:
from datasette.permissions import Resource
# Page size PaginatedResources.all() uses after the first page
ALL_RESOURCES_PAGE_SIZE = 1000
@dataclasses.dataclass
class PaginatedResources:
"""Paginated results from allowed_resources query."""
@ -67,7 +71,10 @@ class PaginatedResources:
for resource in self.resources:
yield resource
# Continue fetching subsequent pages if there are more
# Continue fetching subsequent pages if there are more. Every page
# re-runs the whole permission query, so fetch the rest in the
# largest pages allowed_resources() supports - the caller sees the
# same sequence of resources whatever the page size.
next_token = self.next
while next_token:
page = await self._datasette.allowed_resources(
@ -76,7 +83,7 @@ class PaginatedResources:
parent=self._parent,
include_is_private=self._include_is_private,
include_reasons=self._include_reasons,
limit=self._limit,
limit=max(self._limit, ALL_RESOURCES_PAGE_SIZE),
next=next_token,
)
for resource in page.resources:
@ -851,19 +858,32 @@ def detect_fts_sql(table):
)
_detected_json1 = None
def detect_json1(conn=None):
close_conn = False
"""Does SQLite have the JSON functions? With no ``conn`` this checks the
SQLite library itself, which cannot change while the process runs, so
the answer is cached instead of opening a connection on every call (it
is called for every HTML table page and facet request)."""
global _detected_json1
if conn is None:
conn = sqlite3.connect(":memory:")
close_conn = True
if _detected_json1 is None:
probe = sqlite3.connect(":memory:")
try:
_detected_json1 = _json1_works(probe)
finally:
probe.close()
return _detected_json1
return _json1_works(conn)
def _json1_works(conn):
try:
conn.execute("SELECT json('{}')")
return True
except sqlite3.Error:
return False
finally:
if close_conn:
conn.close()
def table_columns(conn, table):
@ -1212,12 +1232,16 @@ class SpatialiteConnectionProblem(ConnectionProblem):
def check_connection(conn):
tables = [
r[0]
for r in conn.execute(
"select name from sqlite_master where type='table'"
).fetchall()
]
try:
tables = [
r[0]
for r in conn.execute(
"select name from sqlite_master where type='table'"
).fetchall()
]
except sqlite3.DatabaseError as e:
# e.g. "file is not a database"
raise ConnectionProblem(e)
for table in tables:
try:
conn.execute(

336
datasette/utils/catalog.py Normal file
View file

@ -0,0 +1,336 @@
"""Schema facts read from the ``_internal`` catalog instead of the databases.
Pages that summarise many databases at once (the index page, permission
listings) must not open every database: with a capped connection pool a loop
over thousands of databases opens and evicts a connection per database at a
0% hit rate. The SchemaWatcher keeps the ``catalog_*`` tables current (owned
databases from the write path, external ones by polling), so these helpers
answer the same questions as the live introspection methods on
:class:`~datasette.database.Database` with a handful of ``_internal`` queries
that cover any number of databases.
They describe whatever the catalog currently holds - exactly what
``allowed_resources()`` lists - so a listing and the facts used to filter or
decorate it always come from the same snapshot.
"""
import json
from .sqlite import (
_VIRTUAL_TABLE_SHADOW_SUFFIXES,
SQLITE_STAT_TABLES,
_is_fts_content_virtual_table,
_virtual_table_module,
derived_table_dependencies_from_rows,
)
# Tables hidden in databases that contain a geometry_columns table, matching
# Database.hidden_table_names()
SPATIALITE_HIDDEN_TABLES = (
"ElementaryGeometries",
"SpatialIndex",
"geometry_columns",
"spatial_ref_sys",
"spatialite_history",
"sql_statements_log",
"sqlite_sequence",
"views_geometry_columns",
"virts_geometry_columns",
"data_licenses",
"KNN",
"KNN2",
)
def _names_param(names):
return json.dumps(list(names))
def derived_cache_key(state):
"""Identifies one version of a database's catalog rows. A replaced file
can keep its schema_version, so the scan count is part of the key."""
return ("catalog", state.catalog_version, state.stats["scans"])
def remember_derived_table_dependencies(state, table_rows):
"""Called by the SchemaWatcher with the ``(name, sql)`` table rows it has
just written to the catalog, so permission checks never need to read
them back."""
state.db._cached_derived_table_dependencies = (
derived_cache_key(state),
derived_table_dependencies_from_rows(table_rows),
)
async def catalog_derived_table_dependencies(datasette, databases):
"""``{name: {implementation table: source table}}`` for the given watched
``Database`` objects, from the rows in the catalog.
Each database's map is cached against its catalog rows (normally filled
in by the SchemaWatcher when it scans the database); the ones missing -
databases restored from a persisted catalog - are read back in one
``_internal`` query, never from the databases themselves.
"""
result = {}
missing = []
for db in databases:
state = db._watch_state
if state.missing or state.catalog_version is None:
# No catalog rows, so nothing listed that could need filtering
result[db.name] = {}
continue
key = derived_cache_key(state)
cached = db._cached_derived_table_dependencies
if cached is not None and cached[0] == key:
result[db.name] = cached[1]
else:
missing.append((db, key))
if missing:
# Only databases with a virtual table can have dependencies
rows = await datasette.get_internal_database().execute(
"""
select database_name, table_name, sql from catalog_tables
where database_name in (
select database_name from catalog_tables
where coalesce(rootpage, 0) = 0
and database_name in (select value from json_each(:names))
)
order by rowid
""",
{"names": _names_param(db.name for db, _ in missing)},
)
by_database = {}
for database_name, table_name, sql in rows.rows:
by_database.setdefault(database_name, []).append((table_name, sql))
for db, key in missing:
deps = derived_table_dependencies_from_rows(by_database.get(db.name, ()))
db._cached_derived_table_dependencies = (key, deps)
result[db.name] = deps
return result
async def all_derived_table_dependencies(datasette):
"""``{database: dependencies}`` for every watched database, cached until
the catalog next changes (``SchemaWatcher.catalog_generation``), so each
page of ``allowed_resources()`` costs a dictionary lookup instead of a
query against every attached database."""
generation = datasette._schema_watcher.catalog_generation
cached = datasette._catalog_derived_cache
if cached is not None and cached[0] == generation:
return cached[1]
watched = [db for db in datasette.databases.values() if db._watch_state is not None]
result = await catalog_derived_table_dependencies(datasette, watched)
result = {name: deps for name, deps in result.items() if deps}
datasette._catalog_derived_cache = (generation, result)
return result
class CatalogSummary:
"""What the index page needs to know about one database's tables."""
def __init__(self, name):
self.name = name
self.views = set()
# (table, sql, type) for virtual tables and tables typed as shadow
self.special_rows = []
self.has_spatialite = False
self._shadow = None
self._content_fts = None
def _classify(self):
virtual = [(t, sql) for t, sql, _ in self.special_rows if sql is not None]
typed_shadow = {t for t, _, type_ in self.special_rows if type_ == "shadow"}
untyped = any(type_ is None for _, _, type_ in self.special_rows)
shadow = set(typed_shadow)
if untyped:
# Rows written without PRAGMA table_list types: fall back to the
# DDL-based rule for SQLite's built-in modules. Candidate names
# that do not exist are harmless - only existing tables are
# looked up.
for virtual_table, sql in virtual:
module = _virtual_table_module(sql)
for suffix in _VIRTUAL_TABLE_SHADOW_SUFFIXES.get(module, ()):
shadow.add(virtual_table + suffix)
self._shadow = shadow
self._content_fts = {
t for t, sql in virtual if _is_fts_content_virtual_table(sql)
}
def is_hidden(self, table, config_hidden=()):
"""Matches ``table in await db.hidden_table_names()`` for an
existing table."""
if self._shadow is None:
self._classify()
if table in config_hidden:
return True
if table in SQLITE_STAT_TABLES or table.startswith("_"):
return True
if table in self._shadow or table in self._content_fts:
return True
return self.has_spatialite and (
table in SPATIALITE_HIDDEN_TABLES
# sqlite_master "name like 'idx_%'": case-insensitive, "_" is
# a single-character wildcard
or (len(table) >= 4 and table[:3].lower() == "idx")
)
async def catalog_summaries(datasette, database_names):
"""``{database: CatalogSummary}`` for the named databases, in two
``_internal`` queries whatever the number of databases."""
internal = datasette.get_internal_database()
params = {"names": _names_param(database_names)}
summaries = {name: CatalogSummary(name) for name in database_names}
views = await internal.execute(
"""
select database_name, view_name from catalog_views
where database_name in (select value from json_each(:names))
""",
params,
)
for database_name, view_name in views.rows:
summary = summaries.get(database_name)
if summary is not None:
summary.views.add(view_name)
if view_name == "geometry_columns":
summary.has_spatialite = True
special = await internal.execute(
"""
select database_name, table_name,
case when coalesce(rootpage, 0) = 0 then sql end as sql, type
from catalog_tables
where database_name in (select value from json_each(:names))
and (
coalesce(rootpage, 0) = 0
or type is null
or type = 'shadow'
or table_name = 'geometry_columns'
)
order by rowid
""",
params,
)
for database_name, table_name, sql, type_ in special.rows:
summary = summaries.get(database_name)
if summary is None:
continue
if table_name == "geometry_columns":
summary.has_spatialite = True
if sql is not None and _virtual_table_module(sql) is None:
sql = None
if sql is not None or type_ in (None, "shadow"):
summary.special_rows.append((table_name, sql, type_))
return summaries
async def catalog_table_details(datasette, pairs):
"""``{(database, table): {"columns": [...], "primary_keys": [...],
"fts_table": str | None}}`` for the given pairs, matching
``table_columns()``, ``primary_keys()`` and ``fts_table()``."""
from . import detect_fts_sql
pairs = list(pairs)
details = {
pair: {"columns": [], "primary_keys": [], "fts_table": None} for pair in pairs
}
if not pairs:
return details
internal = datasette.get_internal_database()
columns = await internal.execute(
"""
select database_name, table_name, name, is_pk from catalog_columns
where (database_name, table_name) in (
select json_extract(value, '$[0]'), json_extract(value, '$[1]')
from json_each(:pairs)
)
order by database_name, table_name, cid
""",
{"pairs": json.dumps(pairs)},
)
pks = {}
for database_name, table_name, name, is_pk in columns.rows:
entry = details.get((database_name, table_name))
if entry is None:
continue
entry["columns"].append(name)
if is_pk:
pks.setdefault((database_name, table_name), []).append((is_pk, name))
for pair, pk_columns in pks.items():
# Stable sort by position in the primary key, like detect_primary_keys()
pk_columns.sort(key=lambda item: item[0])
details[pair]["primary_keys"] = [name for _, name in pk_columns]
# detect_fts() against the catalog: the same LIKE patterns, so the same
# matching rules; first match in sqlite_master (= catalog rowid) order
fts_args = []
for database_name, table in pairs:
_, fts_params = detect_fts_sql(table)
fts_args.append(
[
database_name,
table,
fts_params["fts_double_quoted"],
fts_params["fts_bracket_quoted"],
]
)
fts = await internal.execute(
r"""
select p.db, p.tbl, (
select c.table_name from catalog_tables c
where c.database_name = p.db
and coalesce(c.rootpage, 0) = 0
and (
c.sql like p.dq escape char(92)
or c.sql like p.bq escape char(92)
or (c.table_name = p.tbl and c.sql like '%VIRTUAL TABLE%USING FTS%')
)
order by c.rowid limit 1
)
from (
select json_extract(value, '$[0]') as db, json_extract(value, '$[1]') as tbl,
json_extract(value, '$[2]') as dq, json_extract(value, '$[3]') as bq
from json_each(:args)
) p
""",
{"args": json.dumps(fts_args)},
)
for database_name, table, fts_table in fts.rows:
entry = details.get((database_name, table))
if entry is not None:
entry["fts_table"] = fts_table
return details
async def catalog_relationship_counts(datasette, database_names):
"""``{database: {table: incoming + outgoing foreign keys}}``, counted the
way ``get_all_foreign_keys()`` builds its lists: compound foreign keys
are skipped, and so are references to tables that do not exist."""
internal = datasette.get_internal_database()
rows = await internal.execute(
"""
with fk as (
select database_name, table_name, id, max("table") as other
from catalog_foreign_keys
where database_name in (select value from json_each(:names))
group by database_name, table_name, id
having count(*) = 1
),
valid as (
select fk.database_name, fk.table_name, fk.other from fk
where exists (
select 1 from catalog_tables t
where t.database_name = fk.database_name and t.table_name = fk.other
)
)
select database_name, table_name, count(*) from (
select database_name, table_name from valid
union all
select database_name, other as table_name from valid
)
group by database_name, table_name
""",
{"names": _names_param(database_names)},
)
counts = {}
for database_name, table_name, n in rows.rows:
counts.setdefault(database_name, {})[table_name] = n
return counts

39
datasette/utils/internal_db.py
View file

@ -4,7 +4,11 @@ from sqlite_utils import Database as SQLiteUtilsDatabase
from sqlite_utils import Migrations
from datasette.utils import escape_sqlite, sqlite3, table_column_details
from datasette.utils.sqlite import supports_table_xinfo
from datasette.utils.sqlite import (
sqlite_table_list_types,
supports_table_xinfo,
table_types_from_rows,
)
INTERNAL_DB_SCHEMA_TABLES = {
"catalog_databases",
@ -201,6 +205,19 @@ def catalog_fingerprint_column(db):
db.execute("ALTER TABLE catalog_databases ADD COLUMN fingerprint TEXT")
@internal_migrations(name="0003_catalog_table_type")
def catalog_table_type_column(db):
# PRAGMA table_list's type for each table ("table", "virtual" or
# "shadow"), so hidden tables can be worked out from the catalog
# without opening the database. NULL means unknown.
if "type" not in db["catalog_tables"].columns_dict:
db.execute("ALTER TABLE catalog_tables ADD COLUMN type TEXT")
# Rows written before this column existed have no type: forget
# their fingerprints so the next startup rescans those databases
# instead of restoring them from the persisted catalog
db.execute("UPDATE catalog_databases SET fingerprint = NULL")
_COLUMNS_SQL = """
SELECT m.name, p.cid, p.name, p.type, p."notnull", p.dflt_value, p.pk, p.hidden
FROM sqlite_master m, pragma_table_xinfo(m.name) p WHERE m.type = 'table'
@ -223,12 +240,22 @@ def collect_schema(conn, database_name):
of three PRAGMA calls per table, falling back to the per-table loop if
that fails (e.g. a virtual table whose module is not loaded).
"""
schema = None
if supports_table_xinfo():
try:
return _collect_schema_joined(conn, database_name)
schema = _collect_schema_joined(conn, database_name)
except sqlite3.DatabaseError:
pass
return _collect_schema_per_table(conn, database_name)
if schema is None:
schema = _collect_schema_per_table(conn, database_name)
# Record each table's type (table/virtual/shadow) as SQLite itself
# reports it - with this connection's extensions loaded, so shadow
# tables of extension modules are recognised too
types = sqlite_table_list_types(conn)
if types is None:
types = table_types_from_rows((t[1], t[3]) for t in schema["tables"])
schema["tables"] = [(*t, types.get(t[1])) for t in schema["tables"]]
return schema
def _collect_schema_joined(conn, database_name):
@ -368,10 +395,10 @@ def write_catalog_entries(conn, entries):
)
conn.executemany(
"""
INSERT INTO catalog_tables (database_name, table_name, rootpage, sql)
values (?, ?, ?, ?)
INSERT INTO catalog_tables (database_name, table_name, rootpage, sql, type)
values (?, ?, ?, ?, ?)
""",
schema["tables"],
[t if len(t) == 5 else (*t, None) for t in schema["tables"]],
)
conn.executemany(
"""

74
datasette/utils/sqlite.py
View file

@ -139,6 +139,11 @@ def check_structured_write_table(conn, table: str, *, allow_missing=False):
raise ValueError("Structured writes require an ordinary table")
SQLITE_STAT_TABLES = frozenset(
{"sqlite_stat1", "sqlite_stat2", "sqlite_stat3", "sqlite_stat4"}
)
def sqlite_hidden_table_names(conn, *, schema: str | None = "main") -> list[str]:
schema_table = _sqlite_schema_table(schema)
try:
@ -147,13 +152,27 @@ def sqlite_hidden_table_names(conn, *, schema: str | None = "main") -> list[str]
).fetchall()
except sqlite3.DatabaseError:
return []
return hidden_table_names_from_rows(
rows,
table_type=lambda name: sqlite_table_type(conn, name, schema=schema),
)
def hidden_table_names_from_rows(rows, table_type) -> list[str]:
"""Hidden tables, given ``(name, sql)`` rows for every table in one
schema (in ``sqlite_master`` order) and a ``table_type(name)`` callable.
Shared by the live introspection above and by the ``_internal`` catalog
(see ``datasette.utils.catalog``), which stores ``PRAGMA table_list``
types when it scans a database.
"""
hidden_tables = []
content_fts_tables = []
for name, sql in rows:
if (
name in {"sqlite_stat1", "sqlite_stat2", "sqlite_stat3", "sqlite_stat4"}
name in SQLITE_STAT_TABLES
or name.startswith("_")
or sqlite_table_type(conn, name, schema=schema) == "shadow"
or table_type(name) == "shadow"
):
hidden_tables.append(name)
elif _is_fts_content_virtual_table(sql):
@ -161,6 +180,48 @@ def sqlite_hidden_table_names(conn, *, schema: str | None = "main") -> list[str]
return sorted(hidden_tables) + content_fts_tables
def table_types_from_rows(rows) -> dict[str, SQLiteTableType]:
"""Classify ``(name, sql)`` table rows as table/virtual/shadow from their
DDL alone - the fallback used when ``PRAGMA table_list`` is unavailable.
Only knows the shadow tables of SQLite's built-in modules."""
rows = list(rows)
names = {name for name, _ in rows}
types = {}
for name, sql in rows:
types[name] = "virtual" if _virtual_table_module(sql) is not None else "table"
for virtual_table, sql in rows:
module = _virtual_table_module(sql)
if module is None:
continue
for suffix in _VIRTUAL_TABLE_SHADOW_SUFFIXES.get(module, ()):
if virtual_table + suffix in names:
types[virtual_table + suffix] = "shadow"
return types
def sqlite_table_list_types(conn, *, schema: str = "main") -> dict | None:
"""``{table name: type}`` from ``PRAGMA table_list`` for one schema, or
None if this SQLite does not support it (before 3.37) or it fails."""
if not supports_table_list():
return None
try:
# The statement form cannot be shadowed by a user-created relation
# called pragma_table_list - see sqlite_table_type()
cursor = conn.execute(f"PRAGMA {_quote_identifier(schema)}.table_list")
columns = [description[0] for description in cursor.description]
types = {}
for row in cursor.fetchall():
record = dict(zip(columns, row))
if record.get("schema") != schema:
continue
row_type = record.get("type")
if row_type in {"table", "virtual", "shadow"}:
types[record.get("name")] = row_type
return types
except sqlite3.DatabaseError:
return None
def sqlite_derived_table_dependencies(
conn, *, schema: str | None = "main"
) -> dict[str, str]:
@ -178,7 +239,16 @@ def sqlite_derived_table_dependencies(
rows = conn.execute(
f"select name, sql from {schema_table} where type = 'table'"
).fetchall()
return derived_table_dependencies_from_rows(rows, schema=schema)
def derived_table_dependencies_from_rows(
rows, *, schema: str | None = "main"
) -> dict[str, str]:
"""``sqlite_derived_table_dependencies()`` for ``(name, sql)`` rows of
every table in one schema - from a live connection or from the
``_internal`` catalog."""
rows = list(rows)
table_names = {row[0] for row in rows}
# SQLite identifiers fold ASCII letters only.
identifier_case = str.maketrans(

13
datasette/views/execute_write.py
View file

@ -209,13 +209,6 @@ async def _create_table_template_sql(datasette, db, actor):
return None
def _analysis_changes_schema(analysis):
return any(
operation.operation in {"create", "alter", "drop"}
for operation in analysis.operations
)
class ExecuteWriteView(BaseView):
name = "execute-write"
has_json_alternate = False
@ -400,8 +393,10 @@ class ExecuteWriteView(BaseView):
status=400,
)
if _analysis_changes_schema(analysis):
await self.ds.refresh_schemas(force=True)
# No refresh_schemas() here: execute_write() already rebuilt this
# database's catalog rows before returning if the SQL changed its
# schema (the SchemaWatcher's post-write check), and a forced
# refresh would stat every attached database on every write
if cursor.rowcount == -1:
message = "Query executed"

171
datasette/views/index.py
View file

@ -9,6 +9,11 @@ from datasette.utils import (
make_slot_function,
)
from datasette.utils.asgi import Response
from datasette.utils.catalog import (
catalog_relationship_counts,
catalog_summaries,
catalog_table_details,
)
from datasette.version import __version__
from .base import BaseView
@ -19,6 +24,15 @@ TRUNCATE_AT = 5
# Only attempt counts if database less than this size in bytes:
COUNT_DB_SIZE_LIMIT = 100 * 1024 * 1024
# Row counts need a count(*) per table in every database listed, so the
# index page only runs them when it lists at most this many databases that
# would need counting. Above that it shows table counts only - plus row
# counts that are already known without opening the database (immutable
# databases with inspect data, or whose counts a page computed earlier).
# Everything else on the page comes from the _internal catalog, so the
# index page never opens a database it is not counting.
COUNT_MAX_DATABASES = 20
class IndexView(BaseView):
name = "index"
@ -32,7 +46,9 @@ class IndexView(BaseView):
"view-database", request.actor, include_is_private=True
)
allowed_databases = [r async for r in db_page.all()]
allowed_db_dict = {r.parent: r for r in allowed_databases}
allowed_db_dict = {
r.parent: r for r in allowed_databases if r.parent in self.ds.databases
}
# Group tables by database
tables_by_db = {}
@ -44,71 +60,57 @@ class IndexView(BaseView):
tables_by_db[t.parent] = {}
tables_by_db[t.parent][t.child] = t
databases = []
# Iterate over allowed databases instead of all databases
for name, allowed_db in allowed_db_dict.items():
db = self.ds.databases[name]
database_private = allowed_db.private
names = list(allowed_db_dict)
summaries = await catalog_summaries(self.ds, names)
all_counts = await self._table_counts(names)
sort_by_relationships = request.args.get("_sort") == "relationships"
need_relationships = {
name for name in names if sort_by_relationships or not all_counts[name]
}
relationship_counts = (
await catalog_relationship_counts(self.ds, need_relationships)
if need_relationships
else {}
)
prepared = []
for name, allowed_db in allowed_db_dict.items():
summary = summaries[name]
# Get allowed tables/views for this database
allowed_for_db = tables_by_db.get(name, {})
# Get table names from allowed set instead of db.table_names()
table_names = [child_name for child_name in allowed_for_db]
hidden_table_names = set(await db.hidden_table_names())
# Determine which allowed items are views
view_names_set = set(await db.view_names())
db_config = self.ds.config.get("databases", {}).get(name, {})
config_hidden = {
t
for t, table_config in db_config.get("tables", {}).items()
if table_config.get("hidden")
}
views = [
{"name": child_name, "private": resource.private}
for child_name, resource in allowed_for_db.items()
if child_name in view_names_set
if child_name in summary.views
]
# Filter to just tables (not views) for table processing
table_names = [name for name in table_names if name not in view_names_set]
# Perform counts only for immutable or DBS with <= COUNT_TABLE_LIMIT tables
table_counts = {}
if not db.is_mutable or db.size < COUNT_DB_SIZE_LIMIT:
table_counts = await db.table_counts(10)
# If any of these are None it means at least one timed out - ignore them all
if any(v is None for v in table_counts.values()):
table_counts = {}
table_counts = all_counts[name]
relationships = relationship_counts.get(name, {})
tables = {}
for table in table_names:
# Check if table is in allowed set
if table not in allowed_for_db:
for table, resource in allowed_for_db.items():
if table in summary.views:
continue
table_columns = await db.table_columns(table)
tables[table] = {
"name": table,
"columns": table_columns,
"primary_keys": await db.primary_keys(table),
"count": table_counts.get(table),
"hidden": table in hidden_table_names,
"fts_table": await db.fts_table(table),
"num_relationships_for_sorting": 0,
"private": allowed_for_db[table].private,
"hidden": summary.is_hidden(table, config_hidden),
"num_relationships_for_sorting": (
relationships.get(table, 0) if name in need_relationships else 0
),
"private": resource.private,
}
if request.args.get("_sort") == "relationships" or not table_counts:
# We will be sorting by number of relationships, so populate that field
all_foreign_keys = await db.get_all_foreign_keys()
for table, foreign_keys in all_foreign_keys.items():
if table in tables:
count = len(foreign_keys["incoming"] + foreign_keys["outgoing"])
tables[table]["num_relationships_for_sorting"] = count
hidden_tables = [t for t in tables.values() if t["hidden"]]
visible_tables = [t for t in tables.values() if not t["hidden"]]
tables_and_views_truncated = list(
sorted(
(t for t in tables.values() if t not in hidden_tables),
visible_tables,
key=lambda t: (
t["num_relationships_for_sorting"],
t["count"] or 0,
@ -117,7 +119,50 @@ class IndexView(BaseView):
reverse=True,
)[:TRUNCATE_AT]
)
prepared.append(
(
name,
allowed_db,
tables,
views,
hidden_tables,
visible_tables,
table_counts,
tables_and_views_truncated,
)
)
# Columns, primary keys and FTS tables, for the tables shown only
details = await catalog_table_details(
self.ds,
[(name, t["name"]) for name, *_, truncated in prepared for t in truncated],
)
databases = []
for (
name,
allowed_db,
tables,
views,
hidden_tables,
visible_tables,
table_counts,
tables_and_views_truncated,
) in prepared:
db = self.ds.databases[name]
tables_and_views_truncated = [
{
"name": t["name"],
"columns": details[(name, t["name"])]["columns"],
"primary_keys": details[(name, t["name"])]["primary_keys"],
"count": t["count"],
"hidden": t["hidden"],
"fts_table": details[(name, t["name"])]["fts_table"],
"num_relationships_for_sorting": t["num_relationships_for_sorting"],
"private": t["private"],
}
for t in tables_and_views_truncated
]
# Only add views if this is less than TRUNCATE_AT
if len(tables_and_views_truncated) < TRUNCATE_AT:
num_views_to_add = TRUNCATE_AT - len(tables_and_views_truncated)
@ -140,7 +185,7 @@ class IndexView(BaseView):
),
"hidden_tables_count": len(hidden_tables),
"views_count": len(views),
"private": database_private,
"private": allowed_db.private,
}
)
@ -189,3 +234,33 @@ class IndexView(BaseView):
"noindex": request.path == "/-/",
},
)
async def _table_counts(self, names):
"""``{database: {table: count}}`` - empty for a database whose
counts are skipped or timed out (see COUNT_MAX_DATABASES)."""
counts = {}
to_count = []
for name in names:
db = self.ds.databases[name]
if not db.is_mutable and db.cached_table_counts is not None:
# Known without opening the database
counts[name] = db.cached_table_counts
else:
counts[name] = {}
to_count.append(db)
if len(to_count) > COUNT_MAX_DATABASES:
return counts
for db in to_count:
if db.is_mutable:
try:
if db.size >= COUNT_DB_SIZE_LIMIT:
continue
except OSError:
# The file has gone: there is nothing to count
continue
table_counts = await db.table_counts(10)
# If any of these are None it means at least one timed out - ignore them all
if any(v is None for v in table_counts.values()):
table_counts = {}
counts[db.name] = table_counts
return counts

114
datasette/views/special.py
View file

@ -18,6 +18,7 @@ from datasette.utils import (
tilde_encode,
)
from datasette.utils.asgi import Forbidden, Response
from datasette.utils.catalog import catalog_summaries, catalog_table_details
from .base import BaseView, View
@ -106,16 +107,27 @@ class AutocompleteDebugView(BaseView):
scanned = 0
reached_scan_limit = False
suggestions = []
for database_name, db in self.ds.databases.items():
# The first 100 tables, in database order then by name, from the
# _internal catalog rather than each database's sqlite_master
names = list(self.ds.databases)
candidates = await self.ds.get_internal_database().execute(
"""
select database_name, table_name
from catalog_tables
join json_each(:names) as j on j.value = catalog_tables.database_name
order by j.key, table_name
limit 100
""",
{"names": json.dumps(names)},
)
by_database = {}
for database_name, table_name in candidates.rows:
by_database.setdefault(database_name, []).append(table_name)
for database_name, table_names in by_database.items():
if scanned >= 100 or len(suggestions) >= 5:
break
remaining = 100 - scanned
results = await db.execute(
"select name from sqlite_master where type = 'table' order by name limit ?",
[remaining],
)
for row in results.rows:
table_name = row["name"]
db = self.ds.databases[database_name]
for table_name in table_names:
scanned += 1
if scanned >= 100:
reached_scan_limit = True
@ -924,24 +936,32 @@ class ApiExplorerView(BaseView):
has_json_alternate = False
async def example_links(self, request):
# Which databases and tables the actor can see, and their columns,
# come from the _internal catalog: building this page must not open
# every attached database
database_page = await self.ds.allowed_resources("view-database", request.actor)
visible_databases = {r.parent async for r in database_page.all()}
table_page = await self.ds.allowed_resources("view-table", request.actor)
visible_tables = {}
async for resource in table_page.all():
visible_tables.setdefault(resource.parent, []).append(resource.child)
names = [name for name in self.ds.databases if name in visible_databases]
summaries = await catalog_summaries(self.ds, names)
mutable_tables = [
(name, table)
for name in names
if self.ds.databases[name].is_mutable
for table in visible_tables.get(name, [])
if table not in summaries[name].views
]
details = await catalog_table_details(self.ds, mutable_tables)
databases = []
for name, db in self.ds.databases.items():
database_visible, _ = await self.ds.check_visibility(
request.actor,
action="view-database",
resource=DatabaseResource(database=name),
)
if not database_visible:
continue
for name in names:
db = self.ds.databases[name]
tables = []
table_names = await db.table_names()
for table in table_names:
visible, _ = await self.ds.check_visibility(
request.actor,
action="view-table",
resource=TableResource(database=name, table=table),
)
if not visible:
for table in visible_tables.get(name, []):
if table in summaries[name].views:
continue
table_links = []
tables.append({"name": table, "links": table_links})
@ -961,7 +981,8 @@ class ApiExplorerView(BaseView):
resource=TableResource(database=name, table=table),
actor=request.actor,
):
pks = await db.primary_keys(table)
pks = details[(name, table)]["primary_keys"]
table_columns = details[(name, table)]["columns"]
table_links.extend(
[
{
@ -972,7 +993,7 @@ class ApiExplorerView(BaseView):
"rows": [
{
column: None
for column in await db.table_columns(table)
for column in table_columns
if column not in pks
}
]
@ -995,7 +1016,7 @@ class ApiExplorerView(BaseView):
)
for column in (
(["rowid"] if not pks else [])
+ await db.table_columns(table)
+ table_columns
)
}
]
@ -1269,15 +1290,16 @@ class SchemaBaseView(BaseView):
has_json_alternate = False
async def get_database_schema(self, database_name, actor):
async def get_database_schema(self, database_name, actor, allowed_table_names=None):
"""Get schema SQL for a database."""
db = self.ds.databases[database_name]
allowed_tables_page = await self.ds.allowed_resources(
"view-table", actor, parent=database_name
)
allowed_table_names = {
resource.child async for resource in allowed_tables_page.all()
}
if allowed_table_names is None:
allowed_tables_page = await self.ds.allowed_resources(
"view-table", actor, parent=database_name
)
allowed_table_names = {
resource.child async for resource in allowed_tables_page.all()
}
result = await db.execute(
"select tbl_name, sql from sqlite_master where sql is not null"
)
@ -1342,10 +1364,30 @@ class InstanceSchemaView(SchemaBaseView):
)
allowed_databases = [r.parent async for r in allowed_databases_page.all()]
# Get schema for each database
# One permission query for every table, not one per database
allowed_tables_page = await self.ds.allowed_resources(
"view-table", request.actor
)
allowed_tables = {}
async for resource in allowed_tables_page.all():
allowed_tables.setdefault(resource.parent, set()).add(resource.child)
# Get schema for each database. This page is a dump of every
# database the actor can see, so it reads each one's sqlite_master
# (the catalog has no index or trigger SQL) - but it skips databases
# with no visible tables or views instead of opening them.
schemas = []
for database_name in allowed_databases:
schema = await self.get_database_schema(database_name, request.actor)
if database_name not in self.ds.databases:
continue
visible = allowed_tables.get(database_name, set())
schema = (
await self.get_database_schema(
database_name, request.actor, allowed_table_names=visible
)
if visible
else ""
)
schemas.append({"database": database_name, "schema": schema})
if format_ == "json":

2
docs/internals.rst
View file

@ -2876,7 +2876,7 @@ The internal database schema is as follows:
database_name TEXT,
table_name TEXT,
rootpage INTEGER,
sql TEXT,
sql TEXT, type TEXT,
PRIMARY KEY (database_name, table_name),
FOREIGN KEY (database_name) REFERENCES catalog_databases(database_name)
);

204
tests/connection_budget.py Normal file
View file

@ -0,0 +1,204 @@
"""Record which databases a block of code touches, and assert a budget.
With a capped, idle-reaped connection pool, anything that loops over every
attached database opens (and evicts) a connection per database at a 0% hit
rate. These helpers make "this request only touches the databases it is
about" a testable property::
async with connection_budget() as budget:
await ds.client.get("/db1/t1.json")
budget.assert_only("db1") # plus _internal, always allowed
What counts as touching a database:
* ``opened``: ``Database.connect()`` calls - pooled read connections, write
connections, isolated connections and the SchemaWatcher's short-lived
scan/pragma connections (``track=False``)
* ``reads``: ``Database._execute_fn()`` calls (every ``execute()``,
``execute_fn()`` and introspection helper), whether or not the pool had an
idle connection - so a warm pool does not hide a loop over every database
* ``writes``: ``execute_write*()``, ``execute_isolated_fn()`` and work run on
the write connection
* ``attached``: databases ATTACHed to a connection while it was prepared
(``--crossdb``)
* ``probes``: ``sqlite3.connect()`` calls that bypass ``Database.connect()``
(``:memory:`` feature probes and the like), keyed by filename
The patches are installed process-wide for the duration of the block, so
background work (the SchemaWatcher's polling task, the read pool's reaper)
that runs during the block is counted too. Tests that want to measure one
request should turn polling off (``schema_watch_interval_ms=0``) or measure
it separately.
"""
import collections
import contextlib
import threading
from datasette import app as app_module
from datasette.database import Database
from datasette.utils import sqlite3
INTERNAL = "__INTERNAL__"
_local = threading.local()
class ConnectionBudget:
def __init__(self):
self._lock = threading.Lock()
self.opened = collections.Counter()
self.reads = collections.Counter()
self.writes = collections.Counter()
self.attached = collections.Counter()
self.probes = collections.Counter()
def _add(self, counter, key):
with self._lock:
counter[key] += 1
def touched(self):
"""Names of every database opened, read, written or ATTACHed."""
names = set()
for counter in (self.opened, self.reads, self.writes, self.attached):
names.update(counter)
return names
def summary(self):
return {
"opened": dict(self.opened),
"reads": dict(self.reads),
"writes": dict(self.writes),
"attached": dict(self.attached),
"probes": dict(self.probes),
}
def assert_only(self, *names, internal=True, probes=0):
"""Assert that nothing outside ``names`` was touched.
``_internal`` is allowed unless ``internal=False``. ``probes`` is the
number of direct ``sqlite3.connect()`` calls allowed.
"""
allowed = set(names)
if internal:
allowed.add(INTERNAL)
extra = self.touched() - allowed
assert not extra, (
f"touched {len(extra)} database(s) outside the budget "
f"{sorted(allowed)}: {sorted(extra)[:10]}"
f"{' ...' if len(extra) > 10 else ''}\n{self.summary()}"
)
total_probes = sum(self.probes.values())
assert (
total_probes <= probes
), f"{total_probes} direct sqlite3.connect() call(s): {dict(self.probes)}"
def assert_at_most(self, n, *, internal=True):
"""Assert that at most ``n`` user databases were touched."""
touched = self.touched()
if internal:
touched.discard(INTERNAL)
assert len(touched) <= n, (
f"touched {len(touched)} databases, budget {n}: "
f"{sorted(touched)[:10]}\n{self.summary()}"
)
def assert_no_connections(self, internal=True):
"""Assert that no connection was opened at all (``_internal`` aside)."""
opened = {
k: v for k, v in self.opened.items() if not (internal and k == INTERNAL)
}
assert not opened, f"opened connections: {opened}"
assert not self.attached, f"attached: {dict(self.attached)}"
assert not self.probes, f"probes: {dict(self.probes)}"
@contextlib.contextmanager
def _patched(budget):
original_connect = Database.connect
original_execute_fn = Database._execute_fn
original_execute_write_fn = Database._execute_write_fn
original_isolated = Database.execute_isolated_fn
original_on_write_conn = Database._execute_on_write_connection
original_prepare = app_module.Datasette._prepare_connection
original_sqlite_connect = sqlite3.connect
def connect(self, *args, **kwargs):
budget._add(budget.opened, self.name)
_local.depth = getattr(_local, "depth", 0) + 1
try:
return original_connect(self, *args, **kwargs)
finally:
_local.depth -= 1
async def _execute_fn(self, fn):
budget._add(budget.reads, self.name)
return await original_execute_fn(self, fn)
async def _execute_write_fn(self, fn, *args, **kwargs):
budget._add(budget.writes, self.name)
return await original_execute_write_fn(self, fn, *args, **kwargs)
async def execute_isolated_fn(self, fn):
budget._add(budget.writes, self.name)
return await original_isolated(self, fn)
async def _execute_on_write_connection(self, fn):
budget._add(budget.writes, self.name)
return await original_on_write_conn(self, fn)
def _prepare_connection(self, conn, database, **kwargs):
result = original_prepare(self, conn, database, **kwargs)
try:
rows = conn.execute("PRAGMA database_list").fetchall()
except Exception: # noqa: BLE001
rows = []
for row in rows:
name = row[1]
if name not in ("main", "temp"):
budget._add(budget.attached, name)
return result
def sqlite_connect(database, *args, **kwargs):
if not getattr(_local, "depth", 0):
budget._add(budget.probes, str(database))
return original_sqlite_connect(database, *args, **kwargs)
Database.connect = connect
Database._execute_fn = _execute_fn
Database._execute_write_fn = _execute_write_fn
Database.execute_isolated_fn = execute_isolated_fn
Database._execute_on_write_connection = _execute_on_write_connection
app_module.Datasette._prepare_connection = _prepare_connection
sqlite3.connect = sqlite_connect
try:
yield budget
finally:
Database.connect = original_connect
Database._execute_fn = original_execute_fn
Database._execute_write_fn = original_execute_write_fn
Database.execute_isolated_fn = original_isolated
Database._execute_on_write_connection = original_on_write_conn
app_module.Datasette._prepare_connection = original_prepare
sqlite3.connect = original_sqlite_connect
class connection_budget:
"""Context manager (sync or async) yielding a ConnectionBudget."""
def __init__(self):
self.budget = ConnectionBudget()
self._cm = None
def __enter__(self):
self._cm = _patched(self.budget)
return self._cm.__enter__()
def __exit__(self, *exc):
return self._cm.__exit__(*exc)
async def __aenter__(self):
return self.__enter__()
async def __aexit__(self, *exc):
return self.__exit__(*exc)

633
tests/test_connection_budget.py Normal file
View file

@ -0,0 +1,633 @@
"""Requests, startup and background work only touch the databases they need.
With a capped, idle-reaped connection pool, a loop over every attached
database opens (and evicts) one connection per database. These tests attach
50 databases and use tests/connection_budget.py to assert which databases
each endpoint opens, reads or writes. Every test uses a fresh instance, so
caches that are only warm after a first request do not hide a sweep.
"""
import json
import os
import shutil
import sqlite3
import time
import pytest
from datasette.app import Datasette
from datasette.utils.catalog import (
catalog_relationship_counts,
catalog_summaries,
catalog_table_details,
)
from datasette.utils.sqlite import (
sqlite_derived_table_dependencies,
supports_table_list,
)
from .connection_budget import INTERNAL, connection_budget
N = 50
@pytest.fixture(autouse=True)
def only_default_plugins():
"""Plugins other tests leave registered (tests/plugins/my_plugin.py and
friends query datasette.get_database() from their hooks) are not part
of what these tests measure: unregister them for the duration."""
from datasette.plugins import DEFAULT_PLUGINS, pm
removed = [
(name, plugin)
for name, plugin in pm.list_name_plugin()
if name not in DEFAULT_PLUGINS
]
for name, _ in removed:
pm.unregister(name=name)
try:
yield
finally:
for name, plugin in removed:
if pm.get_plugin(name) is None:
pm.register(plugin, name=name)
def _make_db(path, i):
conn = sqlite3.connect(path)
conn.executescript("""
create table t2 (id integer primary key, title text);
create table t1 (
id integer primary key, name text, other_id integer references t2(id)
);
""")
conn.executemany(
"insert into t1 (name, other_id) values (?, ?)",
[(f"n{j}", j % 3) for j in range(20)],
)
conn.executemany(
"insert into t2 (title) values (?)", [(f"t{j}",) for j in range(3)]
)
if i % 5 == 0:
conn.executescript("""
create virtual table t1_fts using fts5(
name, content="t1", content_rowid='id'
);
insert into t1_fts(t1_fts) values('rebuild');
""")
if i % 3 == 0:
conn.execute("create view v1 as select * from t1")
if i % 7 == 0:
conn.execute("create table _hidden (x)")
conn.commit()
conn.close()
@pytest.fixture(scope="module")
def db_files(tmp_path_factory):
directory = tmp_path_factory.mktemp("budget")
paths = []
for i in range(N):
path = str(directory / f"db{i:03d}.db")
_make_db(path, i)
paths.append(path)
# Older than the SchemaWatcher's racy window, so a sweep is stat-only
old = time.time() - 60
for path in paths:
os.utime(path, (old, old))
return paths
def _ds(paths, **kwargs):
settings = {"schema_watch_interval_ms": 0}
settings.update(kwargs.pop("settings", {}))
ds = Datasette(paths, settings=settings, **kwargs)
ds.root_enabled = True
return ds
def _root(ds):
return {"ds_actor": ds.client.actor_cookie({"id": "root"})}
async def _get(ds, path, cookies=None):
async with connection_budget() as budget:
response = await ds.client.get(path, cookies=cookies or {})
assert response.status_code == 200, (path, response.status_code, response.text)
return response, budget
@pytest.mark.asyncio
@pytest.mark.parametrize("path", ["/", "/.json", "/-/", "/-/.json"])
@pytest.mark.parametrize("as_root", [False, True])
async def test_index_page_opens_no_database(db_files, path, as_root):
ds = _ds(db_files)
await ds.invoke_startup()
response, budget = await _get(ds, path, _root(ds) if as_root else None)
budget.assert_only()
budget.assert_no_connections()
if path.endswith(".json"):
data = response.json()
assert len(data["databases"]) == N
# More databases than COUNT_MAX_DATABASES: no row counts
assert not any(d["show_table_row_counts"] for d in data["databases"])
db000 = next(d for d in data["databases"] if d["name"] == "db000")
assert db000["tables_count"] == 2 # t1, t2
# _hidden, and t1_fts (an external content FTS table is hidden).
# Its fts5 shadow tables are not listed at all: they derive from
# t1_fts, which is itself derived (the one-hop permission policy)
assert db000["hidden_tables_count"] == 2
assert db000["views_count"] == 1
shown = {t["name"]: t for t in db000["tables_and_views_truncated"]}
assert shown["t1"]["columns"] == ["id", "name", "other_id"]
assert shown["t1"]["primary_keys"] == ["id"]
assert shown["t1"]["fts_table"] == "t1_fts"
# Sorted by foreign key relationships when there are no counts
assert shown["t1"]["num_relationships_for_sorting"] == 1
ds.close()
@pytest.mark.asyncio
async def test_index_page_counts_rows_for_few_databases(db_files):
# At most COUNT_MAX_DATABASES databases: rows are counted, and only the
# databases listed are touched
ds = _ds(db_files[:3])
await ds.invoke_startup()
response, budget = await _get(ds, "/.json")
budget.assert_only("db000", "db001", "db002")
data = response.json()
assert all(d["show_table_row_counts"] for d in data["databases"])
assert data["databases"][0]["table_rows_sum"] == 23
ds.close()
@pytest.mark.asyncio
async def test_index_page_uses_cached_immutable_counts(db_files):
# Immutable databases' counts are computed once, when first needed, and
# reused even when there are too many databases to count
ds = Datasette(
db_files[3:],
immutables=db_files[:3],
settings={"schema_watch_interval_ms": 0},
)
await ds.invoke_startup()
for name in ("db000", "db001", "db002"):
await ds.get_database(name).table_counts()
response, budget = await _get(ds, "/.json")
budget.assert_no_connections()
budget.assert_only()
by_name = {d["name"]: d for d in response.json()["databases"]}
assert by_name["db000"]["show_table_row_counts"]
assert not by_name["db010"]["show_table_row_counts"]
ds.close()
@pytest.mark.asyncio
@pytest.mark.parametrize("path", ["/-/databases", "/-/databases.json"])
async def test_databases_endpoint(db_files, path):
ds = _ds(db_files)
await ds.invoke_startup()
_, budget = await _get(ds, path)
budget.assert_only()
budget.assert_no_connections()
ds.close()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"path",
[
"/db001",
"/db001.json",
"/db000.json?_search=n",
"/db001/t1",
"/db001/t1.json",
"/db001/t1?_facet=name",
"/db001/t1.json?_facet=name",
"/db000/t1_fts.json?_search=n1",
"/db001/-/query?sql=select+1",
"/db001/-/query.json?sql=select+1",
"/db001/-/query",
"/db001/t1/1",
"/db001/t1/1.json",
"/db001/-/schema.json",
"/db001/t1/-/autocomplete?q=n",
],
)
@pytest.mark.parametrize("as_root", [False, True])
async def test_database_scoped_pages_touch_only_their_database(db_files, path, as_root):
ds = _ds(db_files)
await ds.invoke_startup()
database = path.split("/")[1].split(".")[0]
_, budget = await _get(ds, path, _root(ds) if as_root else None)
budget.assert_only(database)
ds.close()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"path,as_root",
[
("/-/versions", False),
("/-/versions.json", False),
("/-/plugins.json", False),
("/-/settings.json", False),
("/-/config.json", False),
("/-/actor.json", False),
("/-/threads", True),
("/-/threads.json", True),
("/-/tasks.json", True),
("/-/actions.json", True),
("/-/jump?q=t1", False),
("/-/jump.json?q=t1", False),
("/-/jump.json", False),
("/-/allowed.json?action=view-table", False),
("/-/allowed.json?action=view-table", True),
("/-/allowed.json?action=view-database", False),
("/-/allowed?action=view-table", True),
("/-/rules.json?action=view-table", True),
("/-/check.json?action=view-table&parent=db001&child=t1", True),
("/-/permissions", True),
("/-/allow-debug", False),
("/-/api", False),
("/-/api", True),
("/-/create-token", True),
("/-/queries.json", False),
("/-/patterns", False),
],
)
async def test_instance_pages_open_no_user_database(db_files, path, as_root):
ds = _ds(db_files)
await ds.invoke_startup()
# /-/versions probes the SQLite library once per instance
await ds.client.get("/-/versions.json")
_, budget = await _get(ds, path, _root(ds) if as_root else None)
budget.assert_only()
budget.assert_no_connections()
ds.close()
@pytest.mark.asyncio
async def test_versions_probes_sqlite_once(db_files):
ds = _ds(db_files[:12], crossdb=True)
await ds.invoke_startup()
async with connection_budget() as budget:
for _ in range(3):
assert (await ds.client.get("/-/versions.json")).status_code == 200
# One :memory: connection for the first request, none after; it is not
# given the --crossdb ATTACHes
assert sum(budget.probes.values()) == 1
assert not budget.attached
budget.assert_only(probes=1)
ds.close()
@pytest.mark.asyncio
async def test_detect_json1_is_cached(db_files):
ds = _ds(db_files[:2])
await ds.invoke_startup()
await ds.client.get("/db001/t1")
_, budget = await _get(ds, "/db001/t1?_facet=name")
assert not budget.probes
ds.close()
@pytest.mark.asyncio
async def test_instance_schema_reads_each_visible_database_once(db_files):
# /-/schema dumps every visible database's sqlite_master: one read each,
# nothing for databases the actor cannot see
ds = _ds(
db_files[:10],
config={"databases": {"db003": {"allow": False}}},
)
await ds.invoke_startup()
_, budget = await _get(ds, "/-/schema.json")
expected = {f"db{i:03d}" for i in range(10)} - {"db003"}
assert budget.touched() - {INTERNAL} == expected
assert all(budget.reads[name] == 1 for name in expected)
ds.close()
@pytest.mark.asyncio
async def test_autocomplete_debug_is_bounded(db_files):
ds = _ds(db_files)
await ds.invoke_startup()
_, budget = await _get(ds, "/-/debug/autocomplete")
# Only databases holding one of the (at most 5) suggestions
budget.assert_at_most(5)
ds.close()
@pytest.mark.asyncio
async def test_execute_write_touches_only_its_database(db_files, tmp_path):
paths = []
for path in db_files[:20]:
target = tmp_path / os.path.basename(path)
shutil.copy(path, target)
paths.append(str(target))
ds = _ds(paths)
await ds.invoke_startup()
async with connection_budget() as budget:
response = await ds.client.post(
"/db004/-/execute-write",
actor={"id": "root"},
data={"sql": "create table new_table (id integer primary key)"},
)
assert response.status_code == 200, response.text
assert "Query executed" in response.text
budget.assert_only("db004")
# The catalog already has the new table
rows = await ds.get_internal_database().execute(
"select 1 from catalog_tables where database_name = 'db004' and table_name = 'new_table'"
)
assert rows.rows
ds.close()
@pytest.mark.asyncio
async def test_telemetry_metric_callbacks_open_nothing(db_files):
from datasette import telemetry
ds = _ds(db_files)
await ds.invoke_startup()
telemetry.register_datasette(ds)
try:
with connection_budget() as budget:
observations = 0
for callback in (
telemetry.observe_sql_thread_limit,
telemetry.observe_sql_thread_queue_depth,
telemetry.observe_pending_queries,
telemetry.observe_write_queue_depth,
telemetry.observe_open_connections,
):
observations += len(list(callback()))
assert observations
budget.assert_only(internal=False)
budget.assert_no_connections(internal=False)
finally:
telemetry.unregister_datasette(ds)
ds.close()
@pytest.mark.asyncio
@pytest.mark.parametrize("num_sql_threads", [0, 3])
async def test_startup_opens_each_database_at_most_once(db_files, num_sql_threads):
imm = db_files[:5]
with connection_budget() as budget:
ds = Datasette(
db_files[5:],
immutables=imm,
settings={
"schema_watch_interval_ms": 0,
"num_sql_threads": num_sql_threads,
},
)
await ds._startup_sequence()
# One catalog scan connection per database, no pooled reads or writes
# (immutable table counts are no longer precomputed at startup)
assert set(budget.opened) - {INTERNAL, "_memory"} == {
f"db{i:03d}" for i in range(N)
}
assert all(
count == 1 for name, count in budget.opened.items() if name != INTERNAL
), budget.opened
assert set(budget.reads) <= {INTERNAL}, budget.reads
assert set(budget.writes) <= {INTERNAL}, budget.writes
assert not budget.attached
ds.close()
@pytest.mark.asyncio
async def test_startup_with_crossdb_attaches_nothing(db_files):
with connection_budget() as budget:
ds = Datasette(
db_files[:12],
settings={"schema_watch_interval_ms": 0},
crossdb=True,
)
await ds._startup_sequence()
assert not budget.attached
# A crossdb query still sees the attached databases
response = await ds.client.get(
"/_memory/-/query.json?sql=select+count(*)+from+db001.t1&_shape=array"
)
assert response.status_code == 200
ds.close()
@pytest.mark.asyncio
async def test_restart_with_persisted_catalog_opens_nothing(db_files, tmp_path):
internal = str(tmp_path / "internal.db")
ds = Datasette(db_files, internal=internal)
await ds.invoke_startup()
ds.close()
with connection_budget() as budget:
ds = Datasette(
db_files, internal=internal, settings={"schema_watch_interval_ms": 0}
)
await ds._startup_sequence()
response = await ds.client.get("/.json")
assert response.status_code == 200
budget.assert_only()
ds.close()
@pytest.mark.asyncio
async def test_idle_sweeps_open_nothing(db_files):
ds = _ds(db_files)
await ds.invoke_startup()
with connection_budget() as budget:
for _ in range(3):
await ds._refresh_schemas(background=True)
await ds.refresh_schemas(force=True)
budget.assert_only(internal=False)
budget.assert_no_connections(internal=False)
ds.close()
@pytest.mark.asyncio
async def test_failed_scan_is_not_retried_by_idle_sweeps(
db_files, tmp_path, monkeypatch
):
from datasette import schema_watcher
# The file was just written: do not wait out the racy window
monkeypatch.setattr(schema_watcher, "RACY_WINDOW_NS", 0)
bad = tmp_path / "bad.db"
bad.write_bytes(b"this is not a SQLite database" * 10)
old = time.time() - 60
os.utime(bad, (old, old))
ds = _ds(db_files[:3] + [str(bad)])
await ds.invoke_startup()
assert ds.get_database("bad")._watch_state.error is not None
with connection_budget() as budget:
for _ in range(3):
await ds._refresh_schemas(background=True)
assert "bad" not in budget.touched()
# ...until the file changes
good = sqlite3.connect(tmp_path / "good.db")
good.execute("create table fixed (id integer primary key)")
good.commit()
good.close()
os.replace(tmp_path / "good.db", bad)
await ds._refresh_schemas(background=True)
rows = await ds.get_internal_database().execute(
"select table_name from catalog_tables where database_name = 'bad'"
)
assert [r[0] for r in rows.rows] == ["fixed"]
ds.close()
@pytest.mark.asyncio
async def test_check_databases_reuses_catalog_scan(db_files, tmp_path):
from datasette.cli import check_databases
from datasette.utils import ConnectionProblem
ds = _ds(db_files)
with connection_budget() as budget:
await check_databases(ds)
# The catalog scan opened each database once; no second pass
assert all(count == 1 for name, count in budget.opened.items() if name != INTERNAL)
assert set(budget.reads) <= {INTERNAL, "_memory"}, budget.reads
ds.close()
bad = tmp_path / "bad.db"
bad.write_bytes(b"this is not a SQLite database" * 10)
ds = _ds(db_files[:3] + [str(bad)])
with pytest.raises(Exception) as excinfo:
await check_databases(ds)
assert "bad.db" in str(excinfo.value)
assert "file is not a database" in str(excinfo.value)
assert not isinstance(excinfo.value, ConnectionProblem)
ds.close()
# Parity: the catalog helpers answer exactly what the live helpers do
@pytest.mark.asyncio
async def test_catalog_helpers_match_live_introspection(tmp_path):
path = str(tmp_path / "misc.db")
conn = sqlite3.connect(path)
conn.executescript("""
create table docs (id integer primary key, title text, body text);
create virtual table docs_fts using fts4(title, body, content="docs");
create virtual table docs_vocab using fts4aux(docs_fts);
create virtual table plain using fts5(x);
create virtual table geo using rtree(id, minx, maxx);
create table _private (x);
create table geometry_columns (f_table_name text);
create table idx_foo_bar (x);
create table IDXA (x);
create table parent (id integer primary key);
create table child (
id integer primary key,
p integer references parent(id),
q integer references missing(id)
);
create table comp (a, b, foreign key (a, b) references parent(id, id));
create table self_ref (id integer primary key, up integer references self_ref(id));
create table wide (b text, a integer, c, primary key (a, b));
create view v_docs as select * from docs;
""")
conn.close()
ds = Datasette(
[path],
config={"databases": {"misc": {"tables": {"docs": {"hidden": True}}}}},
settings={"schema_watch_interval_ms": 0},
)
await ds.invoke_startup()
db = ds.get_database("misc")
tables = await db.table_names()
summary = (await catalog_summaries(ds, ["misc"]))["misc"]
# hidden_table_names() also lists SpatiaLite names that do not exist
hidden = set(await db.hidden_table_names()) & set(tables)
config_hidden = {"docs"}
assert {t for t in tables if summary.is_hidden(t, config_hidden)} == hidden
assert summary.views == set(await db.view_names())
details = await catalog_table_details(ds, [("misc", t) for t in tables])
for table in tables:
assert details[("misc", table)]["columns"] == await db.table_columns(table)
assert details[("misc", table)]["primary_keys"] == await db.primary_keys(table)
assert details[("misc", table)]["fts_table"] == await db.fts_table(table)
relationships = (await catalog_relationship_counts(ds, ["misc"])).get("misc", {})
for table, fks in (await db.get_all_foreign_keys()).items():
assert relationships.get(table, 0) == len(fks["incoming"] + fks["outgoing"])
from datasette.utils.catalog import all_derived_table_dependencies
derived = await all_derived_table_dependencies(ds)
live = await db.execute_fn(sqlite_derived_table_dependencies)
assert derived["misc"] == live
# Restored from the catalog rather than remembered from the scan
db._cached_derived_table_dependencies = None
assert await db.derived_table_dependencies() == live
ds.close()
@pytest.mark.asyncio
@pytest.mark.skipif(not supports_table_list(), reason="needs PRAGMA table_list")
async def test_catalog_records_table_types(tmp_path):
path = str(tmp_path / "types.db")
conn = sqlite3.connect(path)
conn.executescript("""
create table plain (id integer primary key);
create virtual table f using fts5(x);
""")
conn.close()
ds = Datasette([path], settings={"schema_watch_interval_ms": 0})
await ds.invoke_startup()
rows = await ds.get_internal_database().execute(
"select table_name, type from catalog_tables where database_name = 'types'"
)
types = dict(rows.rows)
assert types["plain"] == "table"
assert types["f"] == "virtual"
assert types["f_data"] == "shadow"
ds.close()
@pytest.mark.asyncio
async def test_catalog_derived_dependencies_follow_catalog_changes(tmp_path):
path = str(tmp_path / "live.db")
sqlite3.connect(path).execute("create table docs (body text)").connection.close()
ds = Datasette([path], settings={"schema_watch_interval_ms": 0})
await ds.invoke_startup()
from datasette.utils.catalog import all_derived_table_dependencies
assert "live" not in await all_derived_table_dependencies(ds)
db = ds.get_database("live")
await db.execute_write(
"create virtual table docs_fts using fts5(body, content='docs')"
)
derived = await all_derived_table_dependencies(ds)
assert derived["live"]["docs_fts"] == "docs"
assert derived["live"]["docs_fts_data"] == "docs_fts"
ds.close()
def test_connection_budget_helper_counts(db_files):
# The helper itself: a direct execute_fn is a read, connect() an open
import asyncio
async def run():
ds = _ds(db_files[:2])
await ds.invoke_startup()
db = ds.get_database("db001")
async with connection_budget() as budget:
await db.execute("select 1")
conn = sqlite3.connect(":memory:")
conn.close()
assert budget.reads["db001"] == 1
assert budget.touched() == {"db001"}
with pytest.raises(AssertionError, match="db001"):
budget.assert_only("db000")
ds.close()
return budget
budget = asyncio.run(run())
assert sum(budget.probes.values()) == 1
assert json.dumps(budget.summary())

5
tests/test_fts_permissions.py
View file

@ -95,6 +95,11 @@ async def test_derived_permission_discovery_error_is_retried(monkeypatch, listin
action="view-table", resource=TableResource("data", "documents")
)
# While the catalog is current, dependencies come from its rows and
# nothing is discovered live. Live discovery - the path that can fail -
# happens while a rescan of the database is pending.
db._watch_state.needs_scan = True
db._cached_derived_table_dependencies = None
token = _permission_check_cache.set({})
try:
with monkeypatch.context() as patch:

10
tests/test_internal_db.py
View file

@ -31,6 +31,7 @@ async def test_internal_migrations_recorded(ds_client):
assert [tuple(row) for row in migrations.rows] == [
("datasette_internal", "0001_initial"),
("datasette_internal", "0002_catalog_fingerprint"),
("datasette_internal", "0003_catalog_table_type"),
]
@ -64,6 +65,7 @@ async def test_internal_migrations_adopt_existing_internal_db(tmp_path):
assert [tuple(row) for row in migrations.rows] == [
("datasette_internal", "0001_initial"),
("datasette_internal", "0002_catalog_fingerprint"),
("datasette_internal", "0003_catalog_table_type"),
]
ds.close()
@ -75,7 +77,13 @@ async def test_internal_tables(ds_client):
tables = await internal_db.execute("select * from catalog_tables")
assert len(tables) > 5
table = tables.rows[0]
assert set(table.keys()) == {"rootpage", "table_name", "database_name", "sql"}
assert set(table.keys()) == {
"rootpage",
"table_name",
"database_name",
"sql",
"type",
}
@pytest.mark.asyncio

37
tests/test_lifespan.py
View file

@ -215,18 +215,15 @@ async def test_concurrent_first_requests_all_wait_for_slow_startup():
@pytest.mark.asyncio
async def test_setup_db_still_runs_when_invoke_startup_ran_first(tmp_path, monkeypatch):
# Regression test: `datasette serve` (cli.py _serve_async) calls
# ds.invoke_startup() directly, before uvicorn ever sends a
# lifespan.startup event that drives _startup_sequence(). If
# _startup_sequence()'s fast path only checked `_startup_invoked`, it
# would see startup already done and skip the immutable-database
# table-count precompute (setup_db) entirely - a silent regression
# versus main, where AsgiRunOnFirstRequest ran setup_db unconditionally
# on request #1.
async def test_immutable_table_counts_computed_when_first_needed(tmp_path, monkeypatch):
# Immutable databases used to have their table counts precomputed by
# _startup_sequence(), opening every immutable database at startup.
# They are now computed the first time a page needs them, with the same
# generous time limit, and cached.
db_path = tmp_path / "immutable.db"
conn = sqlite3.connect(str(db_path))
conn.execute("create table t (id integer primary key)")
conn.execute("insert into t default values")
conn.commit()
conn.close()
@ -241,22 +238,22 @@ async def test_setup_db_still_runs_when_invoke_startup_ran_first(tmp_path, monke
monkeypatch.setattr(Database, "table_counts", counting_table_counts)
# Simulate the CLI path: invoke_startup() runs directly and completes
# BEFORE _startup_sequence() ever gets a chance to run setup_db.
await ds.invoke_startup()
await ds._startup_sequence()
assert ds._startup_invoked is True
assert call_count["n"] == 0
# The lifespan/first-request path (or the CLI itself, per the fix)
# calling the shared entry point afterwards must still precompute
# table counts for immutable databases.
await ds._startup_sequence()
assert call_count["n"] == 1
assert ds._setup_db_done is True
assert call_count["n"] == 0
assert ds.get_database("immutable").cached_table_counts is None
# Idempotency: a second call must not recompute.
await ds._startup_sequence()
response = await ds.client.get("/.json")
assert response.json()["databases"][0]["table_rows_sum"] == 1
assert call_count["n"] == 1
assert ds.get_database("immutable").cached_table_counts == {"t": 1}
# Cached from then on
await ds.client.get("/immutable.json")
assert ds.get_database("immutable").cached_table_counts == {"t": 1}
ds.close()
@pytest.mark.asyncio