mirror of
https://github.com/simonw/datasette.git
synced 2026-10-21 00:36:59 +02:00
Remove scratch database APIs from the connection proposal
This commit is contained in:
parent
2bf79d0411
commit
fb2b4a3ad8
13 changed files with 17 additions and 2647 deletions
|
|
@ -55,7 +55,6 @@ from .schema_watcher import (
|
|||
SchemaWatcher,
|
||||
store_closing_fingerprints,
|
||||
)
|
||||
from .scratch import ScratchDatabases
|
||||
from .telemetry import (
|
||||
TelemetryMiddleware,
|
||||
_in_datasette_client,
|
||||
|
|
@ -459,7 +458,6 @@ class Datasette:
|
|||
nolock=False,
|
||||
internal=None,
|
||||
default_deny=False,
|
||||
scratch_dir=None,
|
||||
):
|
||||
self._startup_invoked = False
|
||||
self._shutdown_invoked = False
|
||||
|
|
@ -534,18 +532,6 @@ 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 Datasette created it (or it is empty), else a temporary
|
||||
# directory created on first use. Loaded - and the directory locked
|
||||
# - at the end of __init__, so a failed construction leaves nothing
|
||||
# locked.
|
||||
self._scratch = ScratchDatabases(
|
||||
self,
|
||||
scratch_dir,
|
||||
inferred=scratch_dir is None and config_dir is not None,
|
||||
fallback_dir=(config_dir / "scratch") if config_dir else None,
|
||||
)
|
||||
|
||||
self.cache_headers = cache_headers
|
||||
self._static_asset_hashes = {}
|
||||
self.cors = cors
|
||||
|
|
@ -701,13 +687,6 @@ class Datasette:
|
|||
self.client = DatasetteClient(self)
|
||||
# ds.config is available now: resolve per-database schema_watch modes
|
||||
self._schema_watcher.configure()
|
||||
# Attaches existing scratch databases without opening any of them.
|
||||
# Takes the scratch directory's lock, so nothing after this may fail
|
||||
try:
|
||||
self._scratch.load()
|
||||
except BaseException:
|
||||
self._scratch.close()
|
||||
raise
|
||||
# Last, so metric callbacks never see a partially initialized instance
|
||||
register_datasette(self)
|
||||
|
||||
|
|
@ -1046,7 +1025,7 @@ class Datasette:
|
|||
# --crossdb: each pooled _memory connection ATTACHed the databases
|
||||
# that existed when it was opened. Discard them so the next query
|
||||
# sees databases added since, and none removed since
|
||||
if not self.crossdb or db.is_memory or db.is_scratch:
|
||||
if not self.crossdb or db.is_memory:
|
||||
return
|
||||
memory = self.databases.get("_memory")
|
||||
if memory is not None and memory is not db:
|
||||
|
|
@ -1073,39 +1052,6 @@ class Datasette:
|
|||
self._crossdb_attachments_changed(db)
|
||||
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.
|
||||
|
||||
|
|
@ -1155,13 +1101,6 @@ 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
|
||||
|
||||
|
|
@ -1784,9 +1723,7 @@ class Datasette:
|
|||
if crossdb and self.crossdb and database == "_memory":
|
||||
count = 0
|
||||
for db_name, db in self.databases.items():
|
||||
# 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:
|
||||
if count >= SQLITE_LIMIT_ATTACHED or db.is_memory:
|
||||
continue
|
||||
sql = 'ATTACH DATABASE "file:{path}?{qs}" AS [{name}];'.format(
|
||||
path=db.path,
|
||||
|
|
|
|||
|
|
@ -503,12 +503,6 @@ 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,
|
||||
|
|
@ -543,7 +537,6 @@ def serve(
|
|||
ssl_keyfile,
|
||||
ssl_certfile,
|
||||
internal,
|
||||
scratch_dir=None,
|
||||
return_instance=False,
|
||||
):
|
||||
"""Serve up specified SQLite database files with a web UI"""
|
||||
|
|
@ -611,7 +604,6 @@ def serve(
|
|||
"nolock": nolock,
|
||||
"internal": internal,
|
||||
"default_deny": default_deny,
|
||||
"scratch_dir": scratch_dir,
|
||||
}
|
||||
|
||||
# Separate directories from files
|
||||
|
|
@ -917,8 +909,6 @@ async def check_databases(ds):
|
|||
|
||||
instead of opening every database a second time.
|
||||
|
||||
Scratch databases are never checked: Datasette created them, and a
|
||||
plugin's scratch database should not stop the server from starting.
|
||||
"""
|
||||
await ds._refresh_schemas()
|
||||
to_check = await _databases_needing_check(ds)
|
||||
|
|
@ -945,13 +935,7 @@ async def check_databases(ds):
|
|||
# If --crossdb and more than SQLITE_LIMIT_ATTACHED show warning
|
||||
if (
|
||||
ds.crossdb
|
||||
and len(
|
||||
[
|
||||
db
|
||||
for db in ds.databases.values()
|
||||
if not db.is_memory and not db.is_scratch
|
||||
]
|
||||
)
|
||||
and len([db for db in ds.databases.values() if not db.is_memory])
|
||||
> SQLITE_LIMIT_ATTACHED
|
||||
):
|
||||
msg = f"Warning: --crossdb only works with the first {SQLITE_LIMIT_ATTACHED} attached databases"
|
||||
|
|
@ -960,9 +944,7 @@ async def check_databases(ds):
|
|||
|
||||
async def _databases_needing_check(ds):
|
||||
restored_with_virtual_tables = set()
|
||||
databases = [
|
||||
(name, db) for name, db in list(ds.databases.items()) if not db.is_scratch
|
||||
]
|
||||
databases = [(name, db) for name, db in list(ds.databases.items())]
|
||||
restored = [
|
||||
name
|
||||
for name, db in databases
|
||||
|
|
|
|||
|
|
@ -88,9 +88,6 @@ _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
|
||||
# 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
|
||||
|
|
@ -1373,8 +1370,6 @@ 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:
|
||||
|
|
|
|||
|
|
@ -814,8 +814,8 @@ 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.
|
||||
# It is counted instead, so that deleting a scratch database's files
|
||||
# can wait for it (and raises once the database has been closed).
|
||||
# Counted separately so database lifecycle operations can wait for
|
||||
# scans without closing a connection while its thread is using it.
|
||||
db._untracked_connection_opened()
|
||||
try:
|
||||
conn = db.connect(track=False)
|
||||
|
|
|
|||
1051
datasette/scratch.py
1051
datasette/scratch.py
File diff suppressed because it is too large
Load diff
|
|
@ -140,8 +140,6 @@ 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.
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1395,163 +1395,12 @@ 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 and either is empty or was created by Datasette (it contains ``.datasette-scratch.db``). A ``scratch/`` folder of your own that holds other files is left alone, with a warning in the log. If another Datasette instance is already using it - a server, while you run ``datasette mydir --get ...`` - the second instance starts without its scratch databases, with a warning.
|
||||
- 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 given the same ``--scratch-dir`` fails to start with an error. The first time Datasette opens a directory it adopts any SQLite files already in it, but never deletes anything: leftover ``-wal``, ``-shm`` or ``-journal`` files without their database are only removed once the directory holds Datasette's registry.
|
||||
|
||||
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. ``datasette serve`` does not run its startup check on scratch databases: a scratch database whose file cannot be read is reported in the log and does not stop the server from starting. Pages that list every database, such as the index page, read scratch databases' tables from the catalog and do not open them.
|
||||
|
||||
**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. 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.
|
||||
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.
|
||||
|
||||
``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.
|
||||
|
||||
|
|
|
|||
|
|
@ -56,7 +56,6 @@ 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
|
||||
--------
|
||||
|
|
|
|||
|
|
@ -28,8 +28,7 @@ Checks:
|
|||
on any database file, no write or SQL thread left, the instance itself
|
||||
collected.
|
||||
|
||||
Works against older checkouts too: features they lack (settings, scratch
|
||||
databases, schema watch modes) are detected and skipped.
|
||||
Works against older checkouts too: features they lack (settings and schema watch modes) are detected and skipped.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
|
|
@ -66,10 +65,6 @@ try:
|
|||
from datasette.connection_pool import ConnectionLeaseError
|
||||
except ImportError:
|
||||
ConnectionLeaseError = None
|
||||
try:
|
||||
from datasette import scratch as scratch_module
|
||||
except ImportError:
|
||||
scratch_module = None
|
||||
|
||||
OP_TIMEOUT = 30.0
|
||||
SLOW_SQL = (
|
||||
|
|
@ -125,7 +120,6 @@ SCENARIOS = {
|
|||
"add_remove": 2,
|
||||
"replace_delete": 1,
|
||||
"db_close": 1,
|
||||
"scratch": 3,
|
||||
"cancel_read": 1,
|
||||
},
|
||||
},
|
||||
|
|
@ -148,7 +142,6 @@ SCENARIOS = {
|
|||
"ddl": 1,
|
||||
"isolated": 1,
|
||||
"churn_slow": 1,
|
||||
"scratch": 1,
|
||||
"cancel_read": 1,
|
||||
},
|
||||
},
|
||||
|
|
@ -257,7 +250,6 @@ SCENARIOS = {
|
|||
"churn_read": 2,
|
||||
"add_remove": 1,
|
||||
"replace_delete": 1,
|
||||
"scratch": 2,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
|
@ -370,7 +362,6 @@ class Torture:
|
|||
self.closed_at = None
|
||||
self.stop_at = None
|
||||
self.churn_seq = 0
|
||||
self.scratch_seq = 0
|
||||
self.features = {}
|
||||
self.stable_names = []
|
||||
self.ext_names = []
|
||||
|
|
@ -404,25 +395,15 @@ class Torture:
|
|||
message = str(exc)
|
||||
if isinstance(exc, QueryInterrupted):
|
||||
return "QueryInterrupted"
|
||||
if isinstance(exc, DatasetteClosedError) and (
|
||||
group in ("churn", "scratch") or self.closing
|
||||
):
|
||||
if isinstance(exc, DatasetteClosedError) and (group == "churn" or self.closing):
|
||||
return type(exc).__name__
|
||||
if self.closing and isinstance(exc, RuntimeError):
|
||||
# Work submitted to the executor after close() shut it down
|
||||
return "RuntimeError after close"
|
||||
if scratch_module is not None and isinstance(
|
||||
exc,
|
||||
(
|
||||
scratch_module.ScratchDatabaseNotFound,
|
||||
scratch_module.ScratchDatabaseExists,
|
||||
),
|
||||
):
|
||||
return type(exc).__name__
|
||||
if isinstance(exc, KeyError) and group in ("churn", "scratch"):
|
||||
if isinstance(exc, KeyError) and group == "churn":
|
||||
return "KeyError (database removed)"
|
||||
if isinstance(exc, sqlite3.Error):
|
||||
if group in ("churn", "scratch"):
|
||||
if group == "churn":
|
||||
# Files replaced, deleted and closed underneath
|
||||
return f"sqlite3 on {group}: {message[:40]}"
|
||||
if group == "mem" and "is locked" in message:
|
||||
|
|
@ -501,12 +482,6 @@ class Torture:
|
|||
kwargs = {}
|
||||
if spec.get("crossdb"):
|
||||
kwargs["crossdb"] = True
|
||||
scratch_dir = os.path.join(self.tmp, "scratch")
|
||||
self.features["scratch"] = (
|
||||
"scratch_dir" in inspect.signature(Datasette.__init__).parameters
|
||||
)
|
||||
if self.features["scratch"]:
|
||||
kwargs["scratch_dir"] = scratch_dir
|
||||
self.features["schema_watch"] = (
|
||||
"schema_watch" in inspect.signature(Datasette.add_database).parameters
|
||||
)
|
||||
|
|
@ -940,62 +915,6 @@ class Torture:
|
|||
db.close()
|
||||
self.count("Database.close")
|
||||
|
||||
async def op_scratch(self, w):
|
||||
ds = self.ds
|
||||
if not self.features["scratch"] or self.closing:
|
||||
return
|
||||
scratch = [n for n, g in list(self.groups.items()) if g == "scratch"]
|
||||
roll = w.rng.random()
|
||||
if not scratch or (roll < 0.3 and len(scratch) < 6):
|
||||
with self.lock:
|
||||
self.scratch_seq += 1
|
||||
name = f"s{self.seed}_{self.scratch_seq}"
|
||||
ok, db = await self.call(
|
||||
"scratch_create", "scratch", ds.create_scratch_database(name)
|
||||
)
|
||||
if ok:
|
||||
with self.lock:
|
||||
self.groups[db.name] = "scratch"
|
||||
await self.call(
|
||||
"scratch_write",
|
||||
"scratch",
|
||||
db.execute_write("create table t (id integer primary key, v text)"),
|
||||
)
|
||||
return
|
||||
name = w.rng.choice(scratch)
|
||||
db = self.db(name)
|
||||
if db is None:
|
||||
return
|
||||
if roll < 0.75:
|
||||
if w.rng.random() < 0.5:
|
||||
await self.call(
|
||||
"scratch_write",
|
||||
"scratch",
|
||||
db.execute_write("insert into t (v) values ('x')"),
|
||||
)
|
||||
else:
|
||||
await self.call(
|
||||
"scratch_read", "scratch", db.execute("select count(*) from t")
|
||||
)
|
||||
elif roll < 0.9:
|
||||
with self.lock:
|
||||
self.scratch_seq += 1
|
||||
new_name = f"s{self.seed}_{self.scratch_seq}"
|
||||
ok, new_db = await self.call(
|
||||
"scratch_rename", "scratch", ds.rename_scratch_database(name, new_name)
|
||||
)
|
||||
if ok:
|
||||
with self.lock:
|
||||
self.groups.pop(name, None)
|
||||
self.groups[new_db.name] = "scratch"
|
||||
else:
|
||||
ok, _ = await self.call(
|
||||
"scratch_delete", "scratch", ds.delete_scratch_database(name)
|
||||
)
|
||||
if ok:
|
||||
with self.lock:
|
||||
self.groups.pop(name, None)
|
||||
|
||||
# -- driving ---------------------------------------------------------
|
||||
|
||||
async def worker(self, w, deadline):
|
||||
|
|
@ -1004,7 +923,7 @@ class Torture:
|
|||
weights = [ops[n] for n in names]
|
||||
while time.monotonic() < deadline:
|
||||
op = w.rng.choices(names, weights)[0]
|
||||
if self.closing and op in ("add_remove", "scratch", "replace_delete"):
|
||||
if self.closing and op in ("add_remove", "replace_delete"):
|
||||
op = "read"
|
||||
try:
|
||||
await getattr(self, f"op_{op}")(w)
|
||||
|
|
|
|||
|
|
@ -650,237 +650,3 @@ def test_connection_budget_helper_counts(db_files):
|
|||
budget = asyncio.run(run())
|
||||
assert sum(budget.probes.values()) == 1
|
||||
assert json.dumps(budget.summary())
|
||||
|
||||
|
||||
# Scratch databases: creating, using, listing, renaming and deleting one
|
||||
# touches only that database, and startup with many of them opens each one
|
||||
# at most once (none at all with a persisted catalog)
|
||||
|
||||
|
||||
SCRATCH_N = 1000
|
||||
|
||||
|
||||
def _scratch_probes_only(budget, scratch_dir):
|
||||
"""Direct sqlite3.connect() calls are allowed for the scratch registry
|
||||
and the scratch files themselves (creating a file in WAL mode, the
|
||||
rename checkpoint) - nothing else."""
|
||||
outside = [p for p in budget.probes if not p.startswith(str(scratch_dir))]
|
||||
assert not outside, f"probes outside the scratch directory: {outside}"
|
||||
|
||||
|
||||
def _no_resources_held(db):
|
||||
assert db._all_connections == []
|
||||
assert db._read_pool_state is None
|
||||
assert db._write_thread is None
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def scratch_template_dir(tmp_path_factory):
|
||||
"""SCRATCH_N scratch database files, copied in by hand (adopted at the
|
||||
first startup). Copies of one template, so building it is fast."""
|
||||
directory = tmp_path_factory.mktemp("scratch_template")
|
||||
template = directory / "template.sqlite"
|
||||
conn = sqlite3.connect(template)
|
||||
conn.executescript("""
|
||||
create table t (id integer primary key, v text);
|
||||
insert into t (v) values ('a'), ('b');
|
||||
create view tv as select * from t;
|
||||
""")
|
||||
conn.close()
|
||||
scratch = directory / "scratch"
|
||||
scratch.mkdir()
|
||||
for i in range(SCRATCH_N):
|
||||
shutil.copy(template, scratch / f"s{i:04d}.db")
|
||||
old = time.time() - 60
|
||||
for path in scratch.iterdir():
|
||||
os.utime(path, (old, old))
|
||||
return scratch
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def scratch_1000(scratch_template_dir, tmp_path):
|
||||
target = tmp_path / "scratch"
|
||||
shutil.copytree(scratch_template_dir, target)
|
||||
return target
|
||||
|
||||
|
||||
# Pages that list or check every database (see
|
||||
# test_instance_pages_open_no_user_database and the index tests above)
|
||||
SCRATCH_INSTANCE_PAGES = [
|
||||
("/", False),
|
||||
("/", True),
|
||||
("/.json", False),
|
||||
("/-/databases.json", False),
|
||||
("/-/api", True),
|
||||
("/-/versions.json", False),
|
||||
("/-/jump.json?q=t", False),
|
||||
("/-/allowed.json?action=view-table", False),
|
||||
("/-/allowed.json?action=view-table", True),
|
||||
("/-/allowed.json?action=view-database", False),
|
||||
("/-/check.json?action=view-table&parent=s0001&child=t", True),
|
||||
("/-/permissions", True),
|
||||
]
|
||||
|
||||
|
||||
def _scratch_names(ds):
|
||||
return sorted(name for name, db in ds.databases.items() if db.is_scratch)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("num_sql_threads", [0, 3])
|
||||
async def test_startup_with_many_scratch_databases(scratch_1000, num_sql_threads):
|
||||
settings = {"schema_watch_interval_ms": 0, "num_sql_threads": num_sql_threads}
|
||||
with connection_budget() as budget:
|
||||
ds = Datasette(scratch_dir=str(scratch_1000), settings=settings)
|
||||
# Attached by reading the registry and listing the directory: no
|
||||
# scratch database opened, read or written
|
||||
assert len(_scratch_names(ds)) == SCRATCH_N
|
||||
assert not any(n.startswith("s") for n in budget.touched()), budget.summary()
|
||||
_scratch_probes_only(budget, scratch_1000)
|
||||
assert ds._read_pool_or_none is None
|
||||
|
||||
with connection_budget() as budget:
|
||||
await ds._startup_sequence()
|
||||
# With a temporary internal database each scratch database is opened
|
||||
# once, for its catalog scan - no pooled reads, writes or ATTACH
|
||||
scratch_opened = {k: v for k, v in budget.opened.items() if k.startswith("s")}
|
||||
assert len(scratch_opened) == SCRATCH_N
|
||||
assert set(scratch_opened.values()) == {1}
|
||||
assert set(budget.reads) <= {INTERNAL, "_memory"}, budget.reads
|
||||
assert set(budget.writes) <= {INTERNAL}, budget.writes
|
||||
assert not budget.attached
|
||||
assert not budget.probes, budget.probes
|
||||
for name in _scratch_names(ds)[:: SCRATCH_N // 10]:
|
||||
_no_resources_held(ds.get_database(name))
|
||||
|
||||
# Pages that cover every database, and idle sweeps, open none of them
|
||||
ds.root_enabled = True
|
||||
await ds.client.get("/-/versions.json")
|
||||
with connection_budget() as budget:
|
||||
for path, as_root in SCRATCH_INSTANCE_PAGES:
|
||||
response = await ds.client.get(path, cookies=_root(ds) if as_root else None)
|
||||
assert response.status_code == 200, path
|
||||
await ds._refresh_schemas(background=True)
|
||||
await ds.refresh_schemas(force=True)
|
||||
budget.assert_only("_memory")
|
||||
assert not any(n.startswith("s") for n in budget.opened)
|
||||
data = (await ds.client.get("/.json")).json()
|
||||
s0001 = next(d for d in data["databases"] if d["name"] == "s0001")
|
||||
assert (s0001["tables_count"], s0001["views_count"]) == (1, 1)
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_restart_with_many_scratch_databases_opens_none(scratch_1000, tmp_path):
|
||||
# Not patching RACY_WINDOW_NS: the files were adopted and scanned just
|
||||
# now, and close() records "closed" fingerprints for owned databases,
|
||||
# which are trusted inside the racy window
|
||||
internal = str(tmp_path / "internal.db")
|
||||
settings = {"schema_watch_interval_ms": 0}
|
||||
ds = Datasette(scratch_dir=str(scratch_1000), internal=internal, settings=settings)
|
||||
await ds.invoke_startup()
|
||||
# Written after its catalog scan - still not rescanned at restart
|
||||
await ds.get_database("s0007").execute_write("insert into t (v) values ('c')")
|
||||
ds.close()
|
||||
|
||||
with connection_budget() as budget:
|
||||
ds = Datasette(
|
||||
scratch_dir=str(scratch_1000), internal=internal, settings=settings
|
||||
)
|
||||
await ds._startup_sequence()
|
||||
response = await ds.client.get("/.json")
|
||||
assert response.status_code == 200
|
||||
budget.assert_only("_memory", probes=1)
|
||||
_scratch_probes_only(budget, scratch_1000)
|
||||
assert ds._schema_watcher.counters["restored_from_persisted"] >= SCRATCH_N
|
||||
# The catalog came back without opening them, and they still work
|
||||
rows = await ds.get_internal_database().execute(
|
||||
"select count(*) from catalog_tables where database_name like 's%'"
|
||||
)
|
||||
assert rows.single_value() == SCRATCH_N
|
||||
db = ds.get_database("s0007")
|
||||
assert (await db.execute("select count(*) from t")).single_value() == 3
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_databases_skips_scratch_databases(scratch_1000, tmp_path):
|
||||
from datasette.cli import check_databases
|
||||
|
||||
ds = Datasette(
|
||||
scratch_dir=str(scratch_1000), settings={"schema_watch_interval_ms": 0}
|
||||
)
|
||||
with connection_budget() as budget:
|
||||
await check_databases(ds)
|
||||
# Only the catalog scan (once each): no check_connection() pass over the
|
||||
# scratch databases
|
||||
scratch_opened = {k: v for k, v in budget.opened.items() if k.startswith("s")}
|
||||
assert len(scratch_opened) == SCRATCH_N
|
||||
assert set(scratch_opened.values()) == {1}
|
||||
assert set(budget.reads) <= {INTERNAL, "_memory"}, budget.reads
|
||||
ds.close()
|
||||
|
||||
# A broken file in the scratch directory does not stop the server
|
||||
# starting; it is reported by the catalog scan instead
|
||||
(scratch_1000 / "s0003.db").write_bytes(b"SQLite format 3\x00" + b"x" * 200)
|
||||
ds = Datasette(
|
||||
scratch_dir=str(scratch_1000), settings={"schema_watch_interval_ms": 0}
|
||||
)
|
||||
await check_databases(ds)
|
||||
assert ds.get_database("s0003")._watch_state.error is not None
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("num_sql_threads", [0, 3])
|
||||
async def test_scratch_operations_touch_only_their_database(
|
||||
db_files, tmp_path, num_sql_threads
|
||||
):
|
||||
scratch_dir = tmp_path / "scratch"
|
||||
ds = _ds(
|
||||
db_files,
|
||||
scratch_dir=str(scratch_dir),
|
||||
settings={"num_sql_threads": num_sql_threads},
|
||||
)
|
||||
await ds.invoke_startup()
|
||||
for i in range(10):
|
||||
db = await ds.create_scratch_database(f"other{i}")
|
||||
await db.execute_write("create table t (id integer primary key)")
|
||||
|
||||
async with connection_budget() as budget:
|
||||
db = await ds.create_scratch_database("work", actor={"id": "alice"})
|
||||
budget.assert_only("work", probes=10)
|
||||
_scratch_probes_only(budget, scratch_dir)
|
||||
assert await ds.get_internal_database().execute(
|
||||
"select 1 from catalog_databases where database_name = 'work'"
|
||||
)
|
||||
|
||||
async with connection_budget() as budget:
|
||||
await db.execute_write("create table t (id integer primary key, v text)")
|
||||
await db.execute_write_many(
|
||||
"insert into t (v) values (?)", [(str(i),) for i in range(10)]
|
||||
)
|
||||
response = await ds.client.get("/work/t.json?_shape=array")
|
||||
assert len(response.json()) == 10
|
||||
response = await ds.client.get("/work.json")
|
||||
assert response.status_code == 200
|
||||
budget.assert_only("work")
|
||||
|
||||
async with connection_budget() as budget:
|
||||
infos = await ds.list_scratch_databases()
|
||||
assert len(infos) == 11
|
||||
budget.assert_only(internal=False)
|
||||
budget.assert_no_connections(internal=False)
|
||||
|
||||
async with connection_budget() as budget:
|
||||
db = await ds.rename_scratch_database("work", "renamed")
|
||||
budget.assert_only("work", "renamed", probes=10)
|
||||
_scratch_probes_only(budget, scratch_dir)
|
||||
assert (await db.execute("select count(*) from t")).single_value() == 10
|
||||
|
||||
async with connection_budget() as budget:
|
||||
await ds.delete_scratch_database("renamed")
|
||||
budget.assert_only("renamed", probes=10)
|
||||
_scratch_probes_only(budget, scratch_dir)
|
||||
assert "renamed" not in ds.databases
|
||||
ds.close()
|
||||
|
|
|
|||
|
|
@ -6,7 +6,6 @@ Each test names the problem it pins down.
|
|||
|
||||
import asyncio
|
||||
import gc
|
||||
import logging
|
||||
import os
|
||||
import sqlite3
|
||||
import threading
|
||||
|
|
@ -19,7 +18,6 @@ from datasette import schema_watcher as schema_watcher_module
|
|||
from datasette.app import Datasette
|
||||
from datasette.connection_pool import ConnectionLeaseError
|
||||
from datasette.database import Database
|
||||
from datasette.scratch import LOCK_FILENAME, REGISTRY_FILENAME
|
||||
|
||||
|
||||
def make_db(path, sql="create table t (id integer primary key, v text)"):
|
||||
|
|
@ -290,28 +288,14 @@ def test_block_false_write_from_closed_loop_reaches_catalog(owned_db):
|
|||
assert asyncio.run(later()) == ["added_nonblocking", "t"]
|
||||
|
||||
|
||||
def test_unclosed_datasette_releases_scratch_directory(tmp_path):
|
||||
def test_unclosed_datasette_is_collected():
|
||||
# C4: the temp internal database's atexit.register(bound method) pinned
|
||||
# every Datasette, so the scratch directory lock was never released
|
||||
scratch = tmp_path / "scratch"
|
||||
ds = Datasette(scratch_dir=str(scratch))
|
||||
# every Datasette even after the caller released its last reference.
|
||||
ds = Datasette()
|
||||
ref = weakref.ref(ds)
|
||||
del ds
|
||||
gc.collect()
|
||||
assert ref() is None
|
||||
Datasette(scratch_dir=str(scratch)).close()
|
||||
|
||||
|
||||
def test_failed_construction_releases_scratch_directory(tmp_path):
|
||||
# C4: a Datasette() whose __init__ raised after loading the scratch
|
||||
# directory kept it locked, with no object to close()
|
||||
from datasette.utils import StartupError
|
||||
|
||||
scratch = tmp_path / "scratch"
|
||||
with pytest.raises(StartupError):
|
||||
Datasette(scratch_dir=str(scratch), settings={"default_schema_watch": "bogus"})
|
||||
# No gc.collect(): the lock is taken last, after validation
|
||||
Datasette(scratch_dir=str(scratch)).close()
|
||||
|
||||
|
||||
def test_closed_datasette_is_collected(tmp_path):
|
||||
|
|
@ -498,62 +482,6 @@ async def test_named_memory_database_out_of_band_changes(tmp_path):
|
|||
ds.close()
|
||||
|
||||
|
||||
def _config_dir(tmp_path):
|
||||
config_dir = tmp_path / "cfg"
|
||||
(config_dir / "scratch").mkdir(parents=True)
|
||||
make_db(str(config_dir / "main.db"))
|
||||
return config_dir
|
||||
|
||||
|
||||
def test_config_dir_user_scratch_folder_is_not_taken_over(tmp_path):
|
||||
# Contract C3: an existing config_dir/scratch folder was adopted:
|
||||
# its files were served, orphan sidecars deleted and the folder locked
|
||||
config_dir = _config_dir(tmp_path)
|
||||
make_db(str(config_dir / "scratch" / "private_notes.db"))
|
||||
(config_dir / "scratch" / "other.db-wal").write_bytes(b"junk")
|
||||
ds = Datasette(config_dir=config_dir)
|
||||
assert "private_notes" not in ds.databases
|
||||
assert ds.scratch_dir is None
|
||||
ds.close()
|
||||
assert sorted(os.listdir(config_dir / "scratch")) == [
|
||||
"other.db-wal",
|
||||
"private_notes.db",
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_config_dir_scratch_shared_by_two_instances(tmp_path):
|
||||
# Contract C3: a Datasette-created config_dir/scratch locked out a
|
||||
# second process on the same config dir (datasette cfg --get ...)
|
||||
config_dir = _config_dir(tmp_path)
|
||||
first = Datasette(config_dir=config_dir)
|
||||
assert first.scratch_dir == str((config_dir / "scratch").resolve())
|
||||
await first.create_scratch_database("notes")
|
||||
assert (config_dir / "scratch" / REGISTRY_FILENAME).exists()
|
||||
assert (config_dir / "scratch" / LOCK_FILENAME).exists()
|
||||
second = Datasette(config_dir=config_dir)
|
||||
assert "notes" not in second.databases
|
||||
second.close()
|
||||
first.close()
|
||||
# Once the first has gone, the directory is used again
|
||||
third = Datasette(config_dir=config_dir)
|
||||
assert "notes" in third.databases
|
||||
third.close()
|
||||
|
||||
|
||||
def test_first_use_of_explicit_scratch_dir_keeps_sidecars(tmp_path, caplog):
|
||||
scratch = tmp_path / "scratch"
|
||||
scratch.mkdir()
|
||||
(scratch / "gone.db-wal").write_bytes(b"junk")
|
||||
with caplog.at_level(logging.WARNING, logger="datasette.scratch"):
|
||||
Datasette(scratch_dir=str(scratch)).close()
|
||||
assert (scratch / "gone.db-wal").exists()
|
||||
assert "gone.db-wal" in caplog.text
|
||||
# The next start (the registry exists now) removes it
|
||||
Datasette(scratch_dir=str(scratch)).close()
|
||||
assert not (scratch / "gone.db-wal").exists()
|
||||
|
||||
|
||||
FTS_SETUP = """
|
||||
create table docs (id integer primary key, body text);
|
||||
create virtual table docs_fts using fts5(body, content='docs', content_rowid='id');
|
||||
|
|
@ -757,7 +685,7 @@ async def test_execute_fn_may_return_generator_over_fetched_rows():
|
|||
@pytest.mark.asyncio
|
||||
async def test_index_survives_database_removed_mid_request(tmp_path, monkeypatch):
|
||||
# The index page looked databases up by name again after listing them:
|
||||
# one removed meanwhile (remove_database(), a scratch delete or rename)
|
||||
# one removed meanwhile (remove_database())
|
||||
# failed the whole page with a KeyError
|
||||
from datasette.views import index as index_module
|
||||
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ segfault or a hang fails one test instead of killing the test run.
|
|||
Several event loops (threads) drive one Datasette with extreme settings
|
||||
(smallest pool, 1ms idle timeout, 10ms schema polling) through reads,
|
||||
writes of every kind, DDL, isolated functions, analyze_sql, cancellation,
|
||||
add/remove_database, scratch databases, external schema changes, files
|
||||
add/remove_database, external schema changes, files
|
||||
replaced and deleted, Database.close() and Datasette.close() mid-flight,
|
||||
and non-threaded mode. See the harness docstring for what is checked.
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -1,952 +0,0 @@
|
|||
"""
|
||||
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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue