remote/datasette
1
0
Fork 0
mirror of https://github.com/simonw/datasette.git synced 2026-10-20 16:26:58 +02:00
Code Issues Projects Releases Packages Wiki Activity

Merge spike/scratch into spike/final

Browse source 
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 06:58:02 +00:00
commit 16959f7ff7
No known key found for this signature in database
9 changed files with 2326 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

89
datasette/app.py
View file

@ -49,7 +49,12 @@ from .events import Event
from .plugins import DEFAULT_PLUGINS, get_plugins, pm
from .renderer import json_renderer
from .resources import DatabaseResource, TableResource
from .schema_watcher import FILES_DEFAULT, SchemaWatcher
from .schema_watcher import (
FILES_DEFAULT,
SchemaWatcher,
store_closing_fingerprints,
)
from .scratch import ScratchDatabases
from .telemetry import (
TelemetryMiddleware,
_in_datasette_client,
@ -451,6 +456,7 @@ class Datasette:
nolock=False,
internal=None,
default_deny=False,
scratch_dir=None,
):
self._startup_invoked = False
self._shutdown_invoked = False
@ -526,6 +532,14 @@ class Datasette:
self._internal_database = Database(self, path=internal, mode="rwc")
self._internal_database.name = INTERNAL_DB_NAME
# Scratch databases: an explicit directory, else config_dir/scratch
# if it exists, else a temporary directory created on first use
if scratch_dir is None and config_dir and (config_dir / "scratch").is_dir():
scratch_dir = config_dir / "scratch"
self._scratch = ScratchDatabases(self, scratch_dir)
# Attaches existing scratch databases without opening any of them
self._scratch.load()
self.cache_headers = cache_headers
self._static_asset_hashes = {}
self.cors = cors
@ -989,6 +1003,11 @@ class Datasette:
)
def remove_database(self, name):
db = self._detach_database(name)
db.close()
def _detach_database(self, name):
# remove_database() without the close()
db = self.get_database(name)
new_databases = self.databases.copy()
new_databases.pop(name)
@ -996,7 +1015,40 @@ class Datasette:
# Deletes this database's catalog rows (explicitly - there is no
# periodic stale-catalog scan any more)
self._schema_watcher.unregister(name)
db.close()
return db
@property
def scratch_dir(self):
"""The scratch directory in use, or None if scratch databases are
temporary and none has been created yet."""
return self._scratch.directory
async def create_scratch_database(self, name=None, *, actor=None, metadata=None):
"""Create a new, empty scratch database and attach it.
``name`` defaults to a random ``scratch_xxxxxxxx`` name. ``actor``
(an actor dictionary or an actor ID) is recorded as the owner and
``metadata`` is any JSON-serializable dictionary, both returned by
:meth:`list_scratch_databases`. Returns the
:class:`~datasette.scratch.ScratchDatabase`.
"""
return await self._scratch.create(name, actor=actor, metadata=metadata)
async def delete_scratch_database(self, name):
"""Detach a scratch database and delete its files.
Waits for reads and writes that are already running; calls that are
queued or made later raise
:class:`~datasette.scratch.ScratchDatabaseDeleted`."""
await self._scratch.delete(name)
async def rename_scratch_database(self, name, new_name):
"Rename a scratch database. Returns the new Database object."
return await self._scratch.rename(name, new_name)
async def list_scratch_databases(self):
"List scratch databases as ScratchDatabaseInfo objects, sorted by name."
return await self._scratch.list()
def close(self):
"""Release all resources held by this Datasette instance.
@ -1012,14 +1064,32 @@ class Datasette:
# Stop reporting metrics before closing databases
unregister_datasette(self)
first_exception = None
dbs = list(self.databases.values()) + [self._internal_database]
for db in dbs:
for db in list(self.databases.values()):
try:
db.close()
except Exception as e: # noqa: BLE001
# Collect the first failure and re-raise after every close() has run
if first_exception is None:
first_exception = e
internal = self._internal_database
closing_fingerprints = []
if self.internal_db_created and not internal.is_temp_disk:
# Owned databases are closed now: record their final
# fingerprints so a restart does not rescan them
try:
closing_fingerprints = self._schema_watcher.closing_fingerprints()
except Exception:
logger.exception("Could not fingerprint databases at close")
try:
internal.close()
except Exception as e: # noqa: BLE001
if first_exception is None:
first_exception = e
if closing_fingerprints:
try:
store_closing_fingerprints(internal.path, closing_fingerprints)
except Exception:
logger.exception("Could not store fingerprints at close")
read_pool = self._read_pool_or_none
if read_pool is not None:
read_pool.close()
@ -1029,6 +1099,13 @@ class Datasette:
except Exception as e: # noqa: BLE001
if first_exception is None:
first_exception = e
try:
# Records last_used, unlocks the directory, removes a temporary
# scratch directory
self._scratch.close()
except Exception as e: # noqa: BLE001
if first_exception is None:
first_exception = e
if first_exception is not None:
raise first_exception
@ -1651,7 +1728,9 @@ class Datasette:
if self.crossdb and database == "_memory":
count = 0
for db_name, db in self.databases.items():
if count >= SQLITE_LIMIT_ATTACHED or db.is_memory:
# Scratch databases are never attached: they can be deleted
# while a pooled _memory connection would still hold them
if count >= SQLITE_LIMIT_ATTACHED or db.is_memory or db.is_scratch:
continue
sql = 'ATTACH DATABASE "file:{path}?{qs}" AS [{name}];'.format(
path=db.path,

11
datasette/cli.py
View file

@ -503,6 +503,12 @@ def uninstall(packages, yes):
help="Path to a persistent Datasette internal SQLite database",
envvar="DATASETTE_INTERNAL",
)
@click.option(
"--scratch-dir",
type=click.Path(file_okay=False),
help="Directory for scratch databases, which then survive a restart",
envvar="DATASETTE_SCRATCH_DIR",
)
def serve(
files,
immutable,
@ -537,6 +543,7 @@ def serve(
ssl_keyfile,
ssl_certfile,
internal,
scratch_dir=None,
return_instance=False,
):
"""Serve up specified SQLite database files with a web UI"""
@ -604,6 +611,7 @@ def serve(
"nolock": nolock,
"internal": internal,
"default_deny": default_deny,
"scratch_dir": scratch_dir,
}
# Separate directories from files
@ -899,6 +907,9 @@ 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()):
if database.is_scratch:
# Created by Datasette; checking them would open every one
continue
try:
await database.execute_fn(check_connection)
except SpatialiteConnectionProblem:

39
datasette/database.py
View file

@ -83,6 +83,9 @@ _SHUTDOWN = object()
class Database:
# For table counts stop at this many rows:
count_limit = 10000
# True for ScratchDatabase (datasette.scratch): created and deleted by
# Datasette in its scratch directory
is_scratch = False
def __init__(
self,
@ -156,6 +159,12 @@ class Database:
# connection opened, or the SchemaWatcher stat()ed it). From then on
# write connections never create the file - see connect()
self._file_seen = False
# Short-lived connections that are not in _all_connections (the
# SchemaWatcher's scans and pragma checks): counted, so that deleting
# the file can wait until none is open - see
# _untracked_connection_opened()
self._untracked_open = 0
self._untracked_cond = threading.Condition()
if not is_temp_disk:
self.mode = mode
@ -191,6 +200,32 @@ class Database:
except ValueError:
pass
def _untracked_connection_opened(self):
"""Call before opening a connection with ``connect(track=False)``.
Raises DatasetteClosedError once close() has started, so after
close() no new untracked connection can be opened and
_wait_for_untracked_connections() only has to wait for the ones
already counted."""
with self._untracked_cond:
if self._closed:
raise DatasetteClosedError(f"Database {self.name!r} has been closed")
self._untracked_open += 1
def _untracked_connection_closed(self):
with self._untracked_cond:
self._untracked_open -= 1
if self._untracked_open <= 0:
self._untracked_cond.notify_all()
def _wait_for_untracked_connections(self, timeout=None):
"""Wait until every untracked connection has been closed. Only
meaningful after close(). Returns False on timeout."""
with self._untracked_cond:
return self._untracked_cond.wait_for(
lambda: self._untracked_open <= 0, timeout=timeout
)
def _schema_check(self, conn, loop=None):
# Runs on the thread that owns conn, right after a write task
state = self._watch_state
@ -329,7 +364,7 @@ class Database:
"""
if self._closed:
return
with self._pending_execute_futures_lock:
with self._pending_execute_futures_lock, self._untracked_cond:
if self._closed:
return
self._closed = True
@ -1306,6 +1341,8 @@ class Database:
tags.append("memory")
if self.is_temp_disk:
tags.append("temp_disk")
if self.is_scratch:
tags.append("scratch")
if self.hash:
tags.append(f"hash={self.hash}")
if self.size is not None:

96
datasette/schema_watcher.py
View file

@ -219,17 +219,42 @@ def header_schema_version(path):
return int.from_bytes(header[40:44], "big")
def _fp_to_json(fp, t_ns):
return json.dumps({"fp": fp, "t": t_ns})
def _fp_to_json(fp, t_ns, closed=False):
data = {"fp": fp, "t": t_ns}
if closed:
# Taken by Datasette.close() after every connection to an owned
# database had been closed: nothing in this process can change the
# file after the stat, so the racily-clean rule does not apply
data["closed"] = True
return json.dumps(data)
def _fp_from_json(value):
"""(fp, t_ns, closed) - (None, None, False) if unreadable."""
try:
data = json.loads(value)
fp = tuple(tuple(p) if p is not None else None for p in data["fp"])
return fp, int(data["t"])
return fp, int(data["t"]), bool(data.get("closed"))
except Exception: # noqa: BLE001
return None, None
return None, None, False
def store_closing_fingerprints(internal_path, rows):
"""Write SchemaWatcher.closing_fingerprints() to a persistent internal
database that has already been closed (Datasette.close() is synchronous
and the internal database's write thread is gone by then). Only rows
whose stored schema_version still matches are updated."""
conn = sqlite3.connect(internal_path, isolation_level=None)
try:
conn.execute("BEGIN IMMEDIATE")
conn.executemany(
"UPDATE catalog_databases SET fingerprint = ? "
"WHERE database_name = ? AND schema_version = ?",
[(fp_json, name, version) for name, version, fp_json in rows],
)
conn.execute("COMMIT")
finally:
conn.close()
class SchemaWatcher:
@ -649,20 +674,30 @@ class SchemaWatcher:
# Untracked: Database.close() on the event loop thread must not close
# this connection while a worker thread is using it (that segfaults).
# It is always closed by _close() in the same thread that opened it.
conn = db.connect(track=False)
# It is counted instead, so that deleting a scratch database's files
# can wait for it (and raises once the database has been closed).
db._untracked_connection_opened()
try:
conn = db.connect(track=False)
except BaseException:
db._untracked_connection_closed()
raise
try:
if prepare:
self.ds._prepare_connection(conn, db.name)
else:
conn.row_factory = None
except Exception:
conn.close()
except BaseException:
self._close(db, conn)
raise
return conn
@staticmethod
def _close(db, conn):
conn.close()
try:
conn.close()
finally:
db._untracked_connection_closed()
def _scan_sync(self, state):
"""Read schema_version + full schema in one read transaction."""
@ -804,6 +839,9 @@ class SchemaWatcher:
if state.removed:
continue
if "error" in r:
if state.db._closed:
# Closed (or deleted) while the scan was running
continue
state.error = r["error"]
state.needs_scan = True
logger.warning(
@ -953,8 +991,8 @@ class SchemaWatcher:
continue
if row["path"] != str(state.db.path) or row["schema_version"] is None:
continue
stored_fp, stored_t = _fp_from_json(row["fingerprint"])
if stored_fp is None or is_racy(stored_fp, stored_t):
stored_fp, stored_t, closed = _fp_from_json(row["fingerprint"])
if stored_fp is None or (is_racy(stored_fp, stored_t) and not closed):
continue
t_ns = time.time_ns()
fp = fingerprint(state.db.path)
@ -962,6 +1000,44 @@ class SchemaWatcher:
out[state.name] = (fp, t_ns, row["schema_version"])
return out
# ------------------------------------------------------------------
# shutdown
# ------------------------------------------------------------------
def closing_fingerprints(self):
"""Called by Datasette.close() once every attached database has been
closed. For each owned file database whose catalog is current,
returns ``(name, schema_version, fingerprint_json)`` for the file as
it is now. Owned databases are only changed through Datasette, so a
restart with a persistent internal database can reuse their catalog
rows without opening them - even if data was written (or the -wal
checkpointed by the final close) after the catalog was last built.
Stat calls only."""
out = []
for state in list(self.states.values()):
if (
state.mode != "owned"
or not state.is_file
or state.removed
or state.missing
or state.needs_scan
or state.scan_future is not None
or state.error is not None
or state.catalog_version is None
or state.notified_version not in (None, state.catalog_version)
):
continue
t_ns = time.time_ns()
try:
fp = fingerprint(state.db.path)
except OSError:
continue
if fp[0] is None:
continue
out.append(
(state.name, state.catalog_version, _fp_to_json(fp, t_ns, closed=True))
)
return out
# ------------------------------------------------------------------
# introspection
# ------------------------------------------------------------------

1000
datasette/scratch.py Normal file
View file

File diff suppressed because it is too large Load diff

2
docs/cli-reference.rst
View file

@ -140,6 +140,8 @@ Use ``--internal PATH`` or the ``DATASETTE_INTERNAL`` environment variable to pe
--ssl-certfile TEXT SSL certificate file
--internal PATH Path to a persistent Datasette internal SQLite
database
--scratch-dir DIRECTORY Directory for scratch databases, which then
survive a restart
--help Show this message and exit.

153
docs/internals.rst
View file

@ -1395,12 +1395,163 @@ The ``name`` and ``route`` parameters are optional and work the same way as they
This removes a database that has been previously added. ``name=`` is the unique name of that database.
.. _datasette_scratch_databases:
Scratch databases
-----------------
A *scratch database* is a SQLite file that a plugin (or a user, through a plugin) creates, copies data into, changes, uses for a while and eventually throws away. Scratch databases are served like any other database - ``/name``, the JSON API, the write API - and they survive a restart of Datasette.
They live in the **scratch directory**:
- ``datasette --scratch-dir path/to/scratch`` (or the ``DATASETTE_SCRATCH_DIR`` environment variable, or ``Datasette(scratch_dir=...)``). The directory is created if it does not exist.
- In :ref:`config_dir`, a ``scratch/`` directory inside the configuration directory is used if it exists.
- Otherwise scratch databases are **temporary**: they go in a temporary directory that is created the first time a plugin creates one and deleted when Datasette shuts down, the same way the internal database is temporary unless ``--internal`` is used. ``datasette.scratch_dir`` is ``None`` until then. Plugins can use the same API either way.
Use a directory just for scratch databases. Only one Datasette instance can use a scratch directory at a time: a second one fails to start with an error.
The directory contains one ``<name>.db`` file per scratch database, plus a registry (``.datasette-scratch.db``) that records each database's owner, metadata, creation time and when it was last used. Scratch databases are created in WAL mode (rollback journal mode with ``--nolock``).
**Startup.** Existing scratch databases are attached when Datasette starts without opening any of them: Datasette reads the registry and lists the directory. Like any database, a scratch database gets connections and a write thread only while it is being used, and they are closed again after :ref:`setting_connection_idle_timeout_ms`. With a persistent internal database (``--internal``) and a clean shutdown, the table catalog for scratch databases is restored at startup without opening them either. With the default temporary internal database each scratch database is opened briefly, once, at startup to read its schema - as are, after a crash, the ones that were written to since their schema was last read.
**The catalog** of a scratch database's tables is kept current by its own writes (schema watch mode ``"owned"``): a table created through any of the write methods is in the catalog before the write returns. If another process writes to a scratch database file directly, Datasette will not notice schema changes.
**Copying files by hand.** The directory is the source of truth. A ``.db`` file copied into the scratch directory while Datasette is stopped is adopted as a scratch database at the next startup, with no owner or metadata; its name must follow the naming rules below. A scratch database file deleted by hand is dropped from the registry at the next startup. Do not add, remove or rename files while Datasette is running. Copying a scratch database's ``.db`` file out of the directory gives you an ordinary SQLite database: nothing Datasette-specific is stored inside it.
**Crashes.** Creating, deleting and renaming are recorded in the registry before the files are touched, so if Datasette is killed part-way through, the next startup either finishes the operation (delete, rename) or undoes it (create). It also removes ``-wal``, ``-shm`` and ``-journal`` files whose database file is missing, since SQLite would otherwise apply an old ``-wal`` file to a new database created later under the same name.
**Names** are 1 to 64 characters long, use only letters, digits, ``_`` and ``-``, and start with a letter or a digit. They are unique among all attached databases, and case-insensitively unique among scratch databases (``Foo`` and ``foo`` would be the same file on macOS and Windows).
**Permissions.** Scratch databases are subject to the same permission checks as every other database. Recording an owner does not restrict who can see a scratch database: plugins that create scratch databases for one user should also restrict access to them, for example with the :ref:`plugin_hook_permission_resources_sql` plugin hook.
Scratch databases are not attached to the ``_memory`` database by ``--crossdb``, because they can be deleted at any time. To combine data from several databases in one query, ``ATTACH`` them inside your own write function, as in the recipe below.
.. _datasette_create_scratch_database:
await .create_scratch_database(name=None, \*, actor=None, metadata=None)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
``name`` - string, optional
The name of the new database. Defaults to a random name like ``scratch_3fa85f64``.
``actor`` - dictionary or string, optional
The actor the database is created for. Its ``"id"`` (or the string itself) is recorded as the owner.
``metadata`` - dictionary, optional
Anything JSON-serializable that the plugin wants to remember about this database, returned by :ref:`datasette_list_scratch_databases`.
Creates an empty database file in the scratch directory, attaches it and returns its ``ScratchDatabase`` object, a subclass of :ref:`internals_database` with extra ``.owner``, ``.created``, ``.last_used`` and ``.scratch_metadata`` attributes and ``.is_scratch`` set to ``True``. Raises ``ValueError`` for an invalid name and ``datasette.scratch.ScratchDatabaseExists`` if a database with that name already exists.
.. code-block:: python
db = await datasette.create_scratch_database(
actor=request.actor, metadata={"source": "my-plugin"}
)
await db.execute_write(
"create table notes (id integer primary key, body text)"
)
.. _datasette_delete_scratch_database:
await .delete_scratch_database(name)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Detaches the scratch database and deletes its files (including ``-wal`` and ``-shm``), its catalog entries and its registry entry. Raises ``datasette.scratch.ScratchDatabaseNotFound`` if there is no scratch database with that name.
Reads and writes that are already running when ``delete_scratch_database()`` is called are allowed to finish: the delete waits for them. Reads and writes that are queued but have not started, including ``block=False`` writes, and any call made afterwards using the old ``Database`` object, raise ``datasette.scratch.ScratchDatabaseDeleted`` (a subclass of ``DatasetteClosedError``).
.. _datasette_rename_scratch_database:
await .rename_scratch_database(name, new_name)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Renames a scratch database, keeping its data, owner and metadata, and returns the new ``ScratchDatabase`` object. This detaches the database and waits for running queries in the same way as deleting it, so calls using the old object raise ``ScratchDatabaseDeleted``. Raises ``ScratchDatabaseExists`` if ``new_name`` is taken.
.. _datasette_list_scratch_databases:
await .list_scratch_databases()
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Returns a list of ``datasette.scratch.ScratchDatabaseInfo`` objects, sorted by name, with these attributes:
- ``name`` and ``path``
- ``created`` and ``last_used`` - Unix timestamps. ``last_used`` is the time of the most recent read or write. It is kept in memory and written to the registry about once a minute and when Datasette shuts down, so after a crash it can be up to a minute old.
- ``size`` - in bytes, including the ``-wal`` file
- ``owner`` - the actor ID passed to ``create_scratch_database()``, or ``None``
- ``metadata`` - the dictionary passed to ``create_scratch_database()``
- ``attached`` - ``False`` if the database could not be attached because another database already uses its name
A plugin that cleans up scratch databases nobody has used for a week could do this:
.. code-block:: python
import time
for info in await datasette.list_scratch_databases():
if info.last_used < time.time() - 7 * 24 * 60 * 60:
await datasette.delete_scratch_database(info.name)
.. _datasette_scratch_copy_recipe:
Copying data into a scratch database
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
The fastest way to copy a table into a scratch database is to ``ATTACH`` the source database read-only inside one write function and copy with ``INSERT ... SELECT``. The rows never pass through Python, and 10,000 rows take around 10ms.
.. code-block:: python
from pathlib import Path
from datasette.utils import escape_sqlite
async def copy_table(
datasette, source_name, table, scratch
):
source = datasette.get_database(source_name)
source_uri = Path(source.path).resolve().as_uri() + (
"?mode=ro" if source.is_mutable else "?immutable=1"
)
def copy_in(conn):
conn.execute(
"ATTACH DATABASE ? AS source", [source_uri]
)
try:
with conn:
conn.execute("BEGIN IMMEDIATE")
create_sql = conn.execute(
"select sql from source.sqlite_master "
"where type = 'table' and name = ?",
[table],
).fetchone()[0]
conn.execute(create_sql)
conn.execute(
"insert into main.{t} select * from source.{t}".format(
t=escape_sqlite(table)
)
)
finally:
conn.execute("DETACH DATABASE source")
await scratch.execute_write_fn(
copy_in, transaction=False
)
Notes on this recipe:
- It works for mutable source databases (``?mode=ro``) and immutable ones (``?immutable=1``). It needs a database file: plain in-memory databases cannot be attached this way.
- Pass ``transaction=False`` and manage the transaction yourself, as here. SQLite cannot ``DETACH`` a database that was read inside a transaction that is still open, so with the default ``transaction=True`` the ``DETACH`` fails and the attached database stays on the write connection.
- ``DETACH`` in a ``finally`` block: an ``ATTACH`` lasts as long as the connection, and the write connection is shared by every later write to that database (see :ref:`database_connections`).
- Reusing the original ``CREATE TABLE`` statement keeps the primary key, column types and constraints. ``CREATE TABLE ... AS SELECT`` would lose them.
- With a mutable source the ``INSERT ... SELECT`` reads one consistent snapshot of it. If the source uses a rollback journal rather than WAL, writes to it wait until the copy has finished.
- The table is in the catalog, and visible in Datasette, when ``execute_write_fn()`` returns.
.. _datasette_close:
.close()
--------
Release all resources held by this ``Datasette`` instance. This calls :ref:`database_close` on every attached database (including the internal database), shuts down the thread pool executor used to run SQL queries, and unlinks the temporary file used to back the internal database if one was created.
Release all resources held by this ``Datasette`` instance. This calls :ref:`database_close` on every attached database (including the internal database), shuts down the thread pool executor used to run SQL queries, and unlinks the temporary file used to back the internal database if one was created. It also records when each :ref:`scratch database <datasette_scratch_databases>` was last used and deletes the scratch directory if it was a temporary one.
``close()`` is synchronous, idempotent and one-way: after a call to ``close()`` any attempt to use the Datasette instance to execute SQL will raise a ``datasette.database.DatasetteClosedError`` exception. A closed ``Datasette`` cannot be reopened — callers that need a fresh instance should construct a new one.

1
docs/settings.rst
View file

@ -56,6 +56,7 @@ The files that can be included in this directory are as follows. All are optiona
* ``templates/`` - a directory containing :ref:`customization_custom_templates`
* ``plugins/`` - a directory containing plugins, see :ref:`writing_plugins_one_off`
* ``static/`` - a directory containing static files - these will be served from ``/static/filename.txt``, see :ref:`customization_static_files`
* ``scratch/`` - a directory for :ref:`scratch databases <datasette_scratch_databases>`, used instead of a temporary directory if it exists
Settings
--------

952
tests/test_scratch.py Normal file
View file

@ -0,0 +1,952 @@
"""
Scratch databases: file-based databases that plugins create, fill, change and
delete through the Datasette API, and that survive a restart.
"""
import asyncio
import json
import os
import sqlite3
import subprocess
import sys
import threading
import time
from pathlib import Path
import pytest
from datasette import schema_watcher as schema_watcher_module
from datasette import scratch as scratch_module
from datasette.app import Datasette
from datasette.database import Database, DatasetteClosedError
from datasette.scratch import (
REGISTRY_FILENAME,
ScratchDatabase,
ScratchDatabaseDeleted,
ScratchDatabaseExists,
ScratchDatabaseNotFound,
)
from datasette.utils import StartupError
def _registry_rows(scratch_dir):
conn = sqlite3.connect(os.path.join(scratch_dir, REGISTRY_FILENAME))
try:
conn.row_factory = sqlite3.Row
return {
row["name"]: dict(row)
for row in conn.execute("select * from scratch_databases")
}
finally:
conn.close()
async def _catalog_tables(ds, name):
rows = await ds.get_internal_database().execute(
"select table_name from catalog_tables where database_name = ? order by table_name",
[name],
)
return [r[0] for r in rows.rows]
async def _catalog_has_database(ds, name):
rows = await ds.get_internal_database().execute(
"select 1 from catalog_databases where database_name = ?", [name]
)
return bool(rows.rows)
def _files(scratch_dir, name):
return sorted(f for f in os.listdir(scratch_dir) if f.startswith(name + ".db"))
def _write_threads(name):
return [
t
for t in threading.enumerate()
if t.name == f"_execute_writes for database {name}"
]
def _make_source(path, rows=10000):
conn = sqlite3.connect(path)
conn.execute(
"create table items (id integer primary key, name text not null, score real)"
)
conn.executemany(
"insert into items (name, score) values (?, ?)",
[(f"item {i}", i * 0.5) for i in range(rows)],
)
conn.commit()
conn.close()
return path
@pytest.fixture
def scratch_dir(tmp_path):
return str(tmp_path / "scratch")
# ---------------------------------------------------------------------------
# create / list / names
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_create_list_and_catalog(scratch_dir):
ds = Datasette(scratch_dir=scratch_dir)
await ds.invoke_startup()
db = await ds.create_scratch_database(
"work", actor={"id": "alice"}, metadata={"source": "test"}
)
assert isinstance(db, ScratchDatabase)
assert db.is_scratch
assert ds.get_database("work") is db
assert db.owner == "alice"
assert db.scratch_metadata == {"source": "test"}
assert ds._schema_watcher.states["work"].mode == "owned"
assert os.path.exists(os.path.join(scratch_dir, "work.db"))
# The catalog knows about it before create returns
assert await _catalog_has_database(ds, "work")
# ... and about its tables before each write returns
await db.execute_write("create table t (id integer primary key, v text)")
assert await _catalog_tables(ds, "work") == ["t"]
await db.execute_write("insert into t (v) values ('a')")
# Created in WAL mode
assert (await db.execute("pragma journal_mode")).single_value() == "wal"
infos = await ds.list_scratch_databases()
assert len(infos) == 1
info = infos[0]
assert info.name == "work"
assert info.path == os.path.join(ds.scratch_dir, "work.db")
assert info.owner == "alice"
assert info.metadata == {"source": "test"}
assert info.attached
assert info.size > 0
assert info.created <= info.last_used <= time.time()
assert _registry_rows(scratch_dir)["work"]["state"] == "ready"
# Browsable like any other database
response = await ds.client.get("/work/t.json?_shape=array")
assert response.status_code == 200
assert response.json() == [{"id": 1, "v": "a"}]
ds.close()
@pytest.mark.asyncio
async def test_create_generated_name_and_string_actor(scratch_dir):
ds = Datasette(scratch_dir=scratch_dir)
await ds.invoke_startup()
db1 = await ds.create_scratch_database(actor="bob")
db2 = await ds.create_scratch_database()
assert db1.name != db2.name
assert db1.name.startswith("scratch_")
assert db1.owner == "bob"
assert db2.owner is None
assert [i.name for i in await ds.list_scratch_databases()] == sorted(
[db1.name, db2.name]
)
ds.close()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"name",
[
"",
"../escape",
"a/b",
"a\\b",
".hidden",
"_internal",
"-x",
"a.b",
"a b",
"café",
"x" * 65,
"work\n",
123,
],
)
async def test_invalid_names(scratch_dir, name):
ds = Datasette(scratch_dir=scratch_dir)
with pytest.raises(ValueError):
await ds.create_scratch_database(name)
assert [f for f in os.listdir(scratch_dir) if not f.startswith(".")] == []
ds.close()
@pytest.mark.asyncio
async def test_duplicate_names(scratch_dir, tmp_path):
other = str(tmp_path / "existing.db")
sqlite3.connect(other).execute("vacuum")
ds = Datasette([other], scratch_dir=scratch_dir)
await ds.invoke_startup()
await ds.create_scratch_database("Work")
with pytest.raises(ScratchDatabaseExists):
await ds.create_scratch_database("Work")
# One file on case-insensitive file systems
with pytest.raises(ScratchDatabaseExists):
await ds.create_scratch_database("work")
# Clashes with a database that is not a scratch database
with pytest.raises(ScratchDatabaseExists):
await ds.create_scratch_database("existing")
with pytest.raises(TypeError):
await ds.create_scratch_database("bad_metadata", metadata={"x": object()})
assert "bad_metadata" not in ds.databases
assert sorted(i.name for i in await ds.list_scratch_databases()) == ["Work"]
ds.close()
# ---------------------------------------------------------------------------
# restart
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_restart_attaches_without_opening(scratch_dir, monkeypatch):
ds = Datasette(scratch_dir=scratch_dir)
await ds.invoke_startup()
for i in range(5):
db = await ds.create_scratch_database(
f"s{i}", actor={"id": "alice"}, metadata={"i": i}
)
await db.execute_write("create table t (id integer primary key)")
await db.execute_write("insert into t default values")
ds.close()
connects = []
original_connect = Database.connect
def counting_connect(self, *args, **kwargs):
connects.append(self.name)
return original_connect(self, *args, **kwargs)
monkeypatch.setattr(Database, "connect", counting_connect)
ds = Datasette(scratch_dir=scratch_dir)
# Registered, nothing opened, no threads
assert sorted(n for n in ds.databases if n.startswith("s")) == [
f"s{i}" for i in range(5)
]
assert connects == []
assert ds._read_pool_or_none is None
assert not any(_write_threads(f"s{i}") for i in range(5))
await ds.invoke_startup()
# A temporary internal database has no catalog yet: each scratch
# database was scanned once, on a short-lived connection
assert await _catalog_tables(ds, "s3") == ["t"]
for i in range(5):
db = ds.get_database(f"s{i}")
assert db._all_connections == []
assert db._read_pool_state is None
assert db._write_thread is None
infos = {i.name: i for i in await ds.list_scratch_databases()}
assert infos["s2"].owner == "alice"
assert infos["s2"].metadata == {"i": 2}
assert (
await ds.get_database("s4").execute("select count(*) from t")
).single_value() == 1
ds.close()
@pytest.mark.asyncio
async def test_restart_with_persistent_internal_scans_nothing(
scratch_dir, tmp_path, monkeypatch
):
# Fingerprints stored within the racy window would be rescanned
monkeypatch.setattr(schema_watcher_module, "RACY_WINDOW_NS", 0)
internal = str(tmp_path / "internal.db")
ds = Datasette(scratch_dir=scratch_dir, internal=internal)
await ds.invoke_startup()
for i in range(3):
db = await ds.create_scratch_database(f"s{i}")
await db.execute_write("create table t (id integer primary key)")
await db.execute_write("insert into t default values")
ds.close()
ds = Datasette(scratch_dir=scratch_dir, internal=internal)
await ds.invoke_startup()
watcher = ds._schema_watcher
assert watcher.counters["restored_from_persisted"] == 3
# Data written, and the -wal checkpointed by close(), after the last
# catalog scan: still reused, because close() recorded the final
# fingerprint of each owned database
assert [watcher.states[f"s{i}"].stats["scans"] for i in range(3)] == [0, 0, 0]
assert await _catalog_tables(ds, "s1") == ["t"]
ds.close()
@pytest.mark.asyncio
async def test_directory_lock(scratch_dir):
ds = Datasette(scratch_dir=scratch_dir)
with pytest.raises(StartupError, match="in use by another Datasette instance"):
Datasette(scratch_dir=scratch_dir)
ds.close()
Datasette(scratch_dir=scratch_dir).close()
@pytest.mark.asyncio
async def test_temporary_scratch_dir_when_not_configured():
ds = Datasette()
await ds.invoke_startup()
assert ds.scratch_dir is None
assert not ds._scratch.persistent
db = await ds.create_scratch_database("tmp")
await db.execute_write("create table t (id)")
directory = ds.scratch_dir
assert os.path.isdir(directory)
ds.close()
assert not os.path.exists(directory)
@pytest.mark.asyncio
async def test_config_dir_scratch_subdirectory(tmp_path):
config_dir = tmp_path / "config"
(config_dir / "scratch").mkdir(parents=True)
ds = Datasette(config_dir=config_dir)
assert ds.scratch_dir == str((config_dir / "scratch").resolve())
await ds.create_scratch_database("in_config")
ds.close()
assert (config_dir / "scratch" / "in_config.db").exists()
# Not picked up as a config_dir database (only *.db at the top level)
ds = Datasette(config_dir=config_dir)
assert ds.get_database("in_config").is_scratch
ds.close()
def test_cli_scratch_dir(tmp_path):
from click.testing import CliRunner
from datasette.cli import cli
scratch_dir = str(tmp_path / "scratch")
async def setup():
ds = Datasette(scratch_dir=scratch_dir)
db = await ds.create_scratch_database("from_cli")
await db.execute_write("create table t (id integer primary key)")
await db.execute_write("insert into t default values")
ds.close()
asyncio.run(setup())
result = CliRunner().invoke(
cli,
[
"serve",
"--memory",
"--scratch-dir",
scratch_dir,
"--get",
"/from_cli/t.json?_shape=array",
],
)
assert result.exit_code == 0, result.output
assert json.loads(result.output) == [{"id": 1}]
@pytest.mark.asyncio
async def test_name_clash_at_startup(scratch_dir, tmp_path):
ds = Datasette(scratch_dir=scratch_dir)
await ds.create_scratch_database("clash")
ds.close()
other = str(tmp_path / "clash.db")
sqlite3.connect(other).execute("vacuum")
ds = Datasette([other], scratch_dir=scratch_dir)
assert not ds.get_database("clash").is_scratch
[info] = await ds.list_scratch_databases()
assert info.name == "clash"
assert not info.attached
ds.close()
# ---------------------------------------------------------------------------
# last_used
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_last_used_in_memory_then_flushed_on_close(scratch_dir):
ds = Datasette(scratch_dir=scratch_dir)
db = await ds.create_scratch_database("used")
await db.execute_write("create table t (id)")
transactions = ds._scratch.counters["registry_transactions"]
before = db.last_used
for _ in range(50):
await db.execute("select count(*) from t")
await db.execute_fn(lambda conn: conn.execute("select 1").fetchall())
await db.execute_isolated_fn(lambda conn: None)
assert db.last_used > before
# Never a registry write per read
assert ds._scratch.counters["registry_transactions"] == transactions
in_memory = db.last_used
assert _registry_rows(scratch_dir)["used"]["last_used"] < in_memory
ds.close()
assert _registry_rows(scratch_dir)["used"]["last_used"] == in_memory
ds = Datasette(scratch_dir=scratch_dir)
[info] = await ds.list_scratch_databases()
assert info.last_used == in_memory
ds.close()
@pytest.mark.asyncio
@pytest.mark.parametrize("num_sql_threads", [0, 2])
async def test_last_used_periodic_flush(scratch_dir, monkeypatch, num_sql_threads):
monkeypatch.setattr(scratch_module, "LAST_USED_FLUSH_INTERVAL_S", 0.2)
ds = Datasette(
scratch_dir=scratch_dir, settings={"num_sql_threads": num_sql_threads}
)
db = await ds.create_scratch_database("used")
await db.execute_write("create table t (id)")
flushes = ds._scratch.counters["last_used_flushes"]
deadline = time.monotonic() + 1.1
reads = 0
while time.monotonic() < deadline:
await db.execute("select count(*) from t")
reads += 1
await asyncio.sleep(0.005)
await asyncio.sleep(0.1)
new_flushes = ds._scratch.counters["last_used_flushes"] - flushes
# Several flushes, far fewer than reads
assert 2 <= new_flushes <= 7, (new_flushes, reads)
assert reads > 50
ds.close()
# ---------------------------------------------------------------------------
# delete
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
@pytest.mark.parametrize("num_sql_threads", [0, 3])
async def test_delete_removes_files_catalog_and_registry(scratch_dir, num_sql_threads):
ds = Datasette(
scratch_dir=scratch_dir, settings={"num_sql_threads": num_sql_threads}
)
await ds.invoke_startup()
db = await ds.create_scratch_database("doomed")
await db.execute_write("create table t (id integer primary key, v)")
await db.execute_write_many(
"insert into t (v) values (?)", [(i,) for i in range(100)]
)
assert (await db.execute("select count(*) from t")).single_value() == 100
if num_sql_threads:
# WAL mode: the sidecars exist while connections are open
assert _files(scratch_dir, "doomed") == [
"doomed.db",
"doomed.db-shm",
"doomed.db-wal",
]
assert await _catalog_tables(ds, "doomed") == ["t"]
await ds.delete_scratch_database("doomed")
assert _files(scratch_dir, "doomed") == []
assert "doomed" not in ds.databases
assert "doomed" not in _registry_rows(scratch_dir)
assert not await _catalog_has_database(ds, "doomed")
assert await _catalog_tables(ds, "doomed") == []
assert await ds.list_scratch_databases() == []
with pytest.raises(ScratchDatabaseDeleted, match="'doomed' was deleted"):
await db.execute("select 1")
with pytest.raises(ScratchDatabaseDeleted):
await db.execute_write("insert into t (v) values (1)")
# Still a DatasetteClosedError for code that handles that
with pytest.raises(DatasetteClosedError):
await db.execute_fn(lambda conn: 1)
with pytest.raises(ScratchDatabaseNotFound):
await ds.delete_scratch_database("doomed")
# The name can be used again, for a new empty database
again = await ds.create_scratch_database("doomed")
assert await again.table_names() == []
ds.close()
@pytest.mark.asyncio
async def test_delete_with_reads_and_writes_in_flight(scratch_dir):
# One SQL thread: a slow read occupies it and more reads queue behind it
ds = Datasette(scratch_dir=scratch_dir, settings={"num_sql_threads": 1})
await ds.invoke_startup()
db = await ds.create_scratch_database("busy")
await db.execute_write("create table t (id integer primary key, v)")
await db.execute_write_many(
"insert into t (v) values (?)", [(i,) for i in range(1000)]
)
read_started = threading.Event()
write_started = threading.Event()
def slow_read(conn):
read_started.set()
time.sleep(0.4)
return conn.execute("select count(*) from t").fetchone()[0]
def slow_write(conn):
write_started.set()
time.sleep(0.4)
conn.execute("insert into t (v) values ('slow')")
return "slow write done"
running_read = asyncio.ensure_future(db.execute_fn(slow_read))
running_write = asyncio.ensure_future(db.execute_write_fn(slow_write))
while not (read_started.is_set() and write_started.is_set()):
await asyncio.sleep(0.01)
queued_reads = [
asyncio.ensure_future(db.execute("select count(*) from t")) for _ in range(3)
]
queued_writes = [
asyncio.ensure_future(db.execute_write("insert into t (v) values (1)"))
for _ in range(3)
]
non_blocking = await db.execute_write("insert into t (v) values (2)", block=False)
assert non_blocking is not None
await asyncio.sleep(0.05)
await ds.delete_scratch_database("busy")
# Work that was already running finished normally
assert await running_read == 1000
assert await running_write == "slow write done"
# Work that was queued got a clear error
for future in queued_reads + queued_writes:
with pytest.raises(ScratchDatabaseDeleted, match="'busy' was deleted"):
await future
assert _files(scratch_dir, "busy") == []
ds.close()
@pytest.mark.asyncio
async def test_delete_waits_for_untracked_connections(scratch_dir):
ds = Datasette(scratch_dir=scratch_dir)
await ds.invoke_startup()
db = await ds.create_scratch_database("watched")
await db.execute_write("create table t (id)")
# Simulate a SchemaWatcher scan holding a short-lived connection
watcher = ds._schema_watcher
conn = watcher._connect(db, prepare=False)
conn.execute("select count(*) from t").fetchall()
released = threading.Event()
def release_later():
time.sleep(0.3)
# Still there while the scan connection is open
assert os.path.exists(os.path.join(scratch_dir, "watched.db"))
released.set()
watcher._close(db, conn)
thread = threading.Thread(target=release_later)
thread.start()
start = time.monotonic()
await ds.delete_scratch_database("watched")
assert time.monotonic() - start >= 0.25
assert released.is_set()
thread.join()
assert _files(scratch_dir, "watched") == []
# No new untracked connection once the database is closed
with pytest.raises(DatasetteClosedError):
watcher._connect(db, prepare=False)
ds.close()
@pytest.mark.asyncio
async def test_delete_stress_no_crash(scratch_dir):
"""Delete while reads (with the time-limit progress handler installed)
and writes are running and queued, many times over."""
ds = Datasette(
scratch_dir=scratch_dir,
settings={"num_sql_threads": 3, "sql_time_limit_ms": 2000},
)
await ds.invoke_startup()
for iteration in range(15):
db = await ds.create_scratch_database(f"stress{iteration}")
await db.execute_write("create table t (id integer primary key, v)")
await db.execute_write_many(
"insert into t (v) values (?)", [(i,) for i in range(2000)]
)
calls = []
for i in range(12):
calls.append(
db.execute(
"with recursive c(x) as (select 1 union all select x + 1 from c "
"where x < 20000) select count(*) from c, (select 1 from t limit 3)"
)
)
calls.append(db.execute_write("insert into t (v) values (?)", [i]))
futures = [asyncio.ensure_future(c) for c in calls]
await asyncio.sleep(0.002 * (iteration % 5))
await ds.delete_scratch_database(f"stress{iteration}")
results = await asyncio.gather(*futures, return_exceptions=True)
for result in results:
if isinstance(result, BaseException):
assert isinstance(result, ScratchDatabaseDeleted), repr(result)
assert _files(scratch_dir, f"stress{iteration}") == []
ds.close()
# ---------------------------------------------------------------------------
# rename
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
@pytest.mark.parametrize("num_sql_threads", [0, 3])
async def test_rename(scratch_dir, num_sql_threads):
ds = Datasette(
scratch_dir=scratch_dir, settings={"num_sql_threads": num_sql_threads}
)
await ds.invoke_startup()
db = await ds.create_scratch_database(
"before", actor={"id": "alice"}, metadata={"a": 1}
)
await db.execute_write("create table t (id integer primary key, v)")
await db.execute_write("insert into t (v) values ('kept')")
other = await ds.create_scratch_database("other")
with pytest.raises(ScratchDatabaseExists):
await ds.rename_scratch_database("before", "other")
with pytest.raises(ScratchDatabaseNotFound):
await ds.rename_scratch_database("missing", "x")
with pytest.raises(ValueError):
await ds.rename_scratch_database("before", "../x")
renamed = await ds.rename_scratch_database("before", "after")
assert renamed.name == "after"
assert renamed.owner == "alice"
assert renamed.scratch_metadata == {"a": 1}
assert (await renamed.execute("select v from t")).single_value() == "kept"
assert _files(scratch_dir, "before") == []
assert os.path.exists(os.path.join(scratch_dir, "after.db"))
assert "before" not in ds.databases
assert await _catalog_tables(ds, "before") == []
assert await _catalog_tables(ds, "after") == ["t"]
with pytest.raises(ScratchDatabaseDeleted, match="renamed to 'after'"):
await db.execute("select 1")
rows = _registry_rows(scratch_dir)
assert set(rows) == {"after", "other"}
assert rows["after"]["owner"] == "alice"
# Case-only rename
renamed = await ds.rename_scratch_database("after", "After")
assert (await renamed.execute("select v from t")).single_value() == "kept"
assert sorted(i.name for i in await ds.list_scratch_databases()) == [
"After",
"other",
]
assert other.name == "other"
ds.close()
# ---------------------------------------------------------------------------
# copying data in
# ---------------------------------------------------------------------------
def copy_table_into(scratch_conn, source_db, table):
# The recipe from docs/internals.rst
from datasette.utils import escape_sqlite
mode = "mode=ro" if source_db.is_mutable else "immutable=1"
source_uri = Path(source_db.path).resolve().as_uri() + "?" + mode
scratch_conn.execute("ATTACH DATABASE ? AS source", [source_uri])
try:
with scratch_conn:
scratch_conn.execute("BEGIN IMMEDIATE")
create_sql = scratch_conn.execute(
"select sql from source.sqlite_master where type = 'table' and name = ?",
[table],
).fetchone()[0]
scratch_conn.execute(create_sql)
scratch_conn.execute(
"insert into main.{t} select * from source.{t}".format(
t=escape_sqlite(table)
)
)
finally:
scratch_conn.execute("DETACH DATABASE source")
@pytest.mark.asyncio
@pytest.mark.parametrize("immutable", [False, True])
@pytest.mark.parametrize("num_sql_threads", [0, 3])
async def test_copy_data_in_recipe(tmp_path, scratch_dir, immutable, num_sql_threads):
source_path = _make_source(str(tmp_path / "source.db"))
kwargs = {"immutables": [source_path]} if immutable else {"files": [source_path]}
ds = Datasette(
scratch_dir=scratch_dir,
settings={"num_sql_threads": num_sql_threads},
**kwargs,
)
await ds.invoke_startup()
source = ds.get_database("source")
assert source.is_mutable is not immutable
scratch = await ds.create_scratch_database("copy")
await scratch.execute_write_fn(
lambda conn: copy_table_into(conn, source, "items"), transaction=False
)
assert (await scratch.execute("select count(*) from items")).single_value() == 10000
# Same schema, primary key included
assert await scratch.primary_keys("items") == ["id"]
assert await _catalog_tables(ds, "copy") == ["items"]
# Modify the copy; the source is unchanged
await scratch.execute_write("delete from items where id % 2 = 0")
await scratch.execute_write("update items set score = score * 2")
assert (await scratch.execute("select count(*) from items")).single_value() == 5000
assert (await source.execute("select count(*) from items")).single_value() == 10000
# Nothing left attached to the write connection
attached = await scratch.execute_write_fn(
lambda conn: [r[1] for r in conn.execute("pragma database_list").fetchall()]
)
assert attached == ["main"]
# The alias is free again for the next copy on the same write connection
def attach_again(conn):
conn.execute(
"ATTACH DATABASE ? AS source",
[Path(source.path).resolve().as_uri() + "?mode=ro"],
)
try:
return conn.execute("select count(*) from source.items").fetchone()[0]
finally:
conn.execute("DETACH DATABASE source")
assert await scratch.execute_write_fn(attach_again, transaction=False) == 10000
await ds.delete_scratch_database("copy")
ds.close()
@pytest.mark.asyncio
async def test_detach_inside_managed_transaction_fails(tmp_path, scratch_dir):
# Why the recipe uses transaction=False: SQLite refuses to DETACH a
# database that was read inside the still-open transaction, so the
# attachment would stay on the write connection
source_path = _make_source(str(tmp_path / "source.db"), rows=10)
ds = Datasette([source_path], scratch_dir=scratch_dir)
source = ds.get_database("source")
scratch = await ds.create_scratch_database("tx")
uri = Path(source.path).resolve().as_uri() + "?mode=ro"
def attach_read_detach(conn):
conn.execute("ATTACH DATABASE ? AS source", [uri])
conn.execute("select count(*) from source.items").fetchall()
conn.execute("DETACH DATABASE source")
with pytest.raises(sqlite3.OperationalError, match="database source is locked"):
await scratch.execute_write_fn(attach_read_detach)
# ... and the attachment has leaked into the write connection: the next
# callback on it cannot use the alias
with pytest.raises(sqlite3.OperationalError, match="already in use"):
await scratch.execute_write_fn(
lambda conn: copy_table_into(conn, source, "items"), transaction=False
)
ds.close()
# ---------------------------------------------------------------------------
# crash consistency
# ---------------------------------------------------------------------------
CRASH_SCRIPT = r"""
import asyncio, os, sys
from datasette import scratch
from datasette.app import Datasette
scratch_dir, op, crash_after = sys.argv[1], sys.argv[2], int(sys.argv[3])
steps = [0]
original_write = scratch.ScratchDatabases._registry_write
original_fsync = scratch._fsync_dir
def step():
steps[0] += 1
if steps[0] == crash_after:
os._exit(17)
def registry_write(self, *args, **kwargs):
original_write(self, *args, **kwargs)
step()
def fsync_dir(d):
original_fsync(d)
step()
async def main():
ds = Datasette(scratch_dir=scratch_dir)
await ds.invoke_startup()
if op != "create":
db = await ds.create_scratch_database("victim", actor={"id": "alice"})
await db.execute_write("create table t (id integer primary key)")
await db.execute_write("insert into t default values")
scratch.ScratchDatabases._registry_write = registry_write
scratch._fsync_dir = fsync_dir
if op == "create":
await ds.create_scratch_database("victim", actor={"id": "alice"})
elif op == "delete":
await ds.delete_scratch_database("victim")
elif op == "rename":
await ds.rename_scratch_database("victim", "renamed")
print("completed", steps[0])
os._exit(0)
asyncio.run(main())
"""
@pytest.mark.parametrize("op", ["create", "delete", "rename"])
@pytest.mark.parametrize("crash_after", [1, 2, 3, 4, 99])
def test_crash_consistency(tmp_path, op, crash_after):
"""Kill the process after the Nth registry transaction or directory fsync
of an operation; the next startup must see the old or the new state,
with metadata, and no stray files."""
scratch_dir = str(tmp_path / "scratch")
result = subprocess.run(
[sys.executable, "-c", CRASH_SCRIPT, scratch_dir, op, str(crash_after)],
capture_output=True,
text=True,
timeout=60,
check=False,
)
assert result.returncode in (0, 17), result.stderr
ds = Datasette(scratch_dir=scratch_dir)
infos = {i.name: i for i in asyncio.run(ds.list_scratch_databases())}
files = sorted(f for f in os.listdir(scratch_dir) if not f.startswith("."))
if op == "create" or op == "delete":
assert set(infos) in (set(), {"victim"})
else:
assert set(infos) in ({"victim"}, {"renamed"})
for name, info in infos.items():
assert info.owner == "alice"
assert info.attached
# Only the listed databases and their own -wal/-shm (a process killed
# with connections open leaves those): no orphan file or sidecar
expected = {f"{name}.db" for name in infos}
assert {f for f in files if f.endswith(".db")} == expected
for f in files:
assert f.endswith(".db") or f.rsplit("-", 1)[0] in expected, files
rows = _registry_rows(scratch_dir)
assert set(rows) == set(infos)
assert all(row["state"] == "ready" for row in rows.values())
for name in infos:
db = ds.get_database(name)
tables = asyncio.run(db.table_names())
if op != "create":
assert tables == ["t"]
ds.close()
@pytest.mark.asyncio
async def test_hand_copied_and_hand_deleted_files(scratch_dir, tmp_path):
ds = Datasette(scratch_dir=scratch_dir)
await ds.create_scratch_database("kept", actor={"id": "alice"})
await ds.create_scratch_database("removed_by_hand")
ds.close()
os.unlink(os.path.join(scratch_dir, "removed_by_hand.db"))
# Copied in: a SQLite file, a non-SQLite file and an invalid name
_make_source(os.path.join(scratch_dir, "dropped_in.db"), rows=3)
Path(scratch_dir, "notes.db").write_text("not a database")
_make_source(os.path.join(scratch_dir, "bad name.db"), rows=1)
# A -wal whose database is gone would be replayed into the next file
# of that name
Path(scratch_dir, "ghost.db-wal").write_bytes(b"x" * 100)
ds = Datasette(scratch_dir=scratch_dir)
infos = {i.name: i for i in await ds.list_scratch_databases()}
assert set(infos) == {"kept", "dropped_in"}
assert infos["kept"].owner == "alice"
assert infos["dropped_in"].owner is None
assert (
await ds.get_database("dropped_in").execute("select count(*) from items")
).single_value() == 3
assert not os.path.exists(os.path.join(scratch_dir, "ghost.db-wal"))
# Left alone
assert os.path.exists(os.path.join(scratch_dir, "notes.db"))
assert os.path.exists(os.path.join(scratch_dir, "bad name.db"))
counters = ds._scratch.counters
assert counters["adopted"] == 1
assert counters["dropped_missing"] == 1
assert counters["stray_files_removed"] == 1
ds.close()
# ---------------------------------------------------------------------------
# modes and isolation
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_not_attached_to_crossdb_memory(scratch_dir, tmp_path):
other = str(tmp_path / "regular.db")
sqlite3.connect(other).execute("vacuum")
ds = Datasette([other], scratch_dir=scratch_dir, crossdb=True)
await ds.create_scratch_database("private_scratch")
names = [
r[1]
for r in (await ds.get_database("_memory").execute("pragma database_list")).rows
]
assert "regular" in names
assert "private_scratch" not in names
ds.close()
@pytest.mark.asyncio
async def test_nolock_creates_rollback_journal_databases(scratch_dir):
ds = Datasette(scratch_dir=scratch_dir, nolock=True)
db = await ds.create_scratch_database("nolock")
await db.execute_write("create table t (id)")
assert (await db.execute("pragma journal_mode")).single_value() == "delete"
assert (await db.execute("select count(*) from t")).single_value() == 0
ds.close()
NO_THREADS_SCRIPT = r"""
import asyncio, os, sys, threading
def no_threads(self):
raise RuntimeError("can't start new thread")
threading.Thread.start = no_threads
from datasette.app import Datasette
scratch_dir = sys.argv[1]
async def main():
ds = Datasette(scratch_dir=scratch_dir, settings={"num_sql_threads": 0})
await ds.invoke_startup()
db = await ds.create_scratch_database("nothreads", actor={"id": "a"})
await db.execute_write("create table t (id integer primary key)")
await db.execute_write("insert into t default values")
assert (await db.execute("select count(*) from t")).single_value() == 1
renamed = await ds.rename_scratch_database("nothreads", "renamed")
assert [i.name for i in await ds.list_scratch_databases()] == ["renamed"]
await ds.delete_scratch_database("renamed")
assert [f for f in os.listdir(scratch_dir) if not f.startswith(".")] == []
ds.close()
print("ok")
asyncio.run(main())
"""
def test_no_threads(tmp_path):
# Pyodide: nothing in the scratch database lifecycle may need a thread
result = subprocess.run(
[sys.executable, "-c", NO_THREADS_SCRIPT, str(tmp_path / "scratch")],
capture_output=True,
text=True,
timeout=60,
check=False,
)
assert result.returncode == 0, result.stdout + result.stderr
assert result.stdout.strip() == "ok"
@pytest.mark.asyncio
async def test_idle_scratch_database_releases_everything(scratch_dir):
ds = Datasette(
scratch_dir=scratch_dir, settings={"connection_idle_timeout_ms": 200}
)
await ds.invoke_startup()
db = await ds.create_scratch_database("idle")
await db.execute_write("create table t (id)")
await db.execute("select count(*) from t")
assert db._write_thread is not None
for _ in range(100):
await asyncio.sleep(0.05)
if (
db._write_thread is None
and not db._retiring_write_threads
and db._all_connections == []
):
break
assert db._write_thread is None
assert db._all_connections == []
assert not _write_threads("idle")
ds.close()