diff --git a/datasette/app.py b/datasette/app.py index e090a4f1..c3ef4d90 100644 --- a/datasette/app.py +++ b/datasette/app.py @@ -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() diff --git a/datasette/cli.py b/datasette/cli.py index e83de93a..8d11eefb 100644 --- a/datasette/cli.py +++ b/datasette/cli.py @@ -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 diff --git a/datasette/database.py b/datasette/database.py index ef60bc57..b816b26e 100644 --- a/datasette/database.py +++ b/datasette/database.py @@ -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 diff --git a/datasette/schema_watcher.py b/datasette/schema_watcher.py index d7c47bbf..3c8333f6 100644 --- a/datasette/schema_watcher.py +++ b/datasette/schema_watcher.py @@ -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 = {} diff --git a/datasette/utils/__init__.py b/datasette/utils/__init__.py index 669d04c7..c36f8a58 100644 --- a/datasette/utils/__init__.py +++ b/datasette/utils/__init__.py @@ -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( diff --git a/datasette/utils/catalog.py b/datasette/utils/catalog.py new file mode 100644 index 00000000..d5d1e4ec --- /dev/null +++ b/datasette/utils/catalog.py @@ -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 diff --git a/datasette/utils/internal_db.py b/datasette/utils/internal_db.py index 57f27314..a71d732c 100644 --- a/datasette/utils/internal_db.py +++ b/datasette/utils/internal_db.py @@ -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( """ diff --git a/datasette/utils/sqlite.py b/datasette/utils/sqlite.py index 2ae1be9b..b5d9cfae 100644 --- a/datasette/utils/sqlite.py +++ b/datasette/utils/sqlite.py @@ -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( diff --git a/datasette/views/execute_write.py b/datasette/views/execute_write.py index c4f5e3fe..4de43493 100644 --- a/datasette/views/execute_write.py +++ b/datasette/views/execute_write.py @@ -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" diff --git a/datasette/views/index.py b/datasette/views/index.py index f73ee38a..bb01fd61 100644 --- a/datasette/views/index.py +++ b/datasette/views/index.py @@ -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 diff --git a/datasette/views/special.py b/datasette/views/special.py index 421dc523..d8c8aa3f 100644 --- a/datasette/views/special.py +++ b/datasette/views/special.py @@ -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": diff --git a/docs/internals.rst b/docs/internals.rst index 4312372a..3e5ccad2 100644 --- a/docs/internals.rst +++ b/docs/internals.rst @@ -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) ); diff --git a/tests/connection_budget.py b/tests/connection_budget.py new file mode 100644 index 00000000..3cc6d906 --- /dev/null +++ b/tests/connection_budget.py @@ -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) diff --git a/tests/test_connection_budget.py b/tests/test_connection_budget.py new file mode 100644 index 00000000..8ecb00f0 --- /dev/null +++ b/tests/test_connection_budget.py @@ -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()) diff --git a/tests/test_fts_permissions.py b/tests/test_fts_permissions.py index 24a40cef..99044c3a 100644 --- a/tests/test_fts_permissions.py +++ b/tests/test_fts_permissions.py @@ -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: diff --git a/tests/test_internal_db.py b/tests/test_internal_db.py index 76ccd148..c3c08716 100644 --- a/tests/test_internal_db.py +++ b/tests/test_internal_db.py @@ -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 diff --git a/tests/test_lifespan.py b/tests/test_lifespan.py index 3a25ccbc..606c20db 100644 --- a/tests/test_lifespan.py +++ b/tests/test_lifespan.py @@ -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