remote/datasette
1
0
Fork 0
mirror of https://github.com/simonw/datasette.git synced 2026-05-29 21:26:59 +02:00
Code Issues Projects Releases Packages Wiki Activity

Refactor hidden_table_names() to use new implemenatation

Browse source 
Refs https://github.com/simonw/datasette/pull/2749#issuecomment-4565727978
This commit is contained in:
Simon Willison 2026-05-28 08:42:06 -07:00
commit aaf00e9ec2
4 changed files with 43 additions and 87 deletions
Download patch file Download diff file Expand all files Collapse all files

80
datasette/database.py
View file

@ -26,7 +26,7 @@ from .utils import (
table_column_details,
)
from .utils.sql_analysis import SQLAnalysis, analyze_sql_tables
from .utils.sqlite import sqlite_version
from .utils.sqlite import sqlite_hidden_table_names
from .inspect import inspect_hash
connections = threading.local()
@ -702,83 +702,7 @@ class Database:
t for t in db_config["tables"] if db_config["tables"][t].get("hidden")
]
if sqlite_version()[1] >= 37:
hidden_tables += [x[0] for x in await self.execute("""
with shadow_tables as (
select name
from pragma_table_list
where [type] = 'shadow'
order by name
),
core_tables as (
select name
from sqlite_master
WHERE name in ('sqlite_stat1', 'sqlite_stat2', 'sqlite_stat3', 'sqlite_stat4')
OR substr(name, 1, 1) == '_'
),
combined as (
select name from shadow_tables
union all
select name from core_tables
)
select name from combined order by 1
""")]
else:
hidden_tables += [x[0] for x in await self.execute("""
WITH base AS (
SELECT name
FROM sqlite_master
WHERE name IN ('sqlite_stat1', 'sqlite_stat2', 'sqlite_stat3', 'sqlite_stat4')
OR substr(name, 1, 1) == '_'
),
fts_suffixes AS (
SELECT column1 AS suffix
FROM (VALUES ('_data'), ('_idx'), ('_docsize'), ('_content'), ('_config'))
),
fts5_names AS (
SELECT name
FROM sqlite_master
WHERE sql LIKE '%VIRTUAL TABLE%USING FTS%'
),
fts5_shadow_tables AS (
SELECT
printf('%s%s', fts5_names.name, fts_suffixes.suffix) AS name
FROM fts5_names
JOIN fts_suffixes
),
fts3_suffixes AS (
SELECT column1 AS suffix
FROM (VALUES ('_content'), ('_segdir'), ('_segments'), ('_stat'), ('_docsize'))
),
fts3_names AS (
SELECT name
FROM sqlite_master
WHERE sql LIKE '%VIRTUAL TABLE%USING FTS3%'
OR sql LIKE '%VIRTUAL TABLE%USING FTS4%'
),
fts3_shadow_tables AS (
SELECT
printf('%s%s', fts3_names.name, fts3_suffixes.suffix) AS name
FROM fts3_names
JOIN fts3_suffixes
),
final AS (
SELECT name FROM base
UNION ALL
SELECT name FROM fts5_shadow_tables
UNION ALL
SELECT name FROM fts3_shadow_tables
)
SELECT name FROM final ORDER BY 1
""")]
# Also hide any FTS tables that have a content= argument
hidden_tables += [x[0] for x in await self.execute("""
SELECT name
FROM sqlite_master
WHERE sql LIKE '%VIRTUAL TABLE%'
AND sql LIKE '%USING FTS%'
AND sql LIKE '%content=%'
""")]
hidden_tables += await self.execute_fn(sqlite_hidden_table_names)
has_spatialite = await self.execute_fn(detect_spatialite)
if has_spatialite:

29
datasette/utils/sqlite.py
View file

@ -80,6 +80,28 @@ def sqlite_table_type(
return _sqlite_table_type_from_schema(conn, table, schema=schema)
def sqlite_hidden_table_names(conn, *, schema: str | None = "main") -> list[str]:
schema_table = _sqlite_schema_table(schema)
try:
rows = conn.execute(
"select name, sql from {} where type = 'table'".format(schema_table)
).fetchall()
except sqlite3.DatabaseError:
return []
hidden_tables = []
content_fts_tables = []
for name, sql in rows:
if (
name in {"sqlite_stat1", "sqlite_stat2", "sqlite_stat3", "sqlite_stat4"}
or name.startswith("_")
or sqlite_table_type(conn, name, schema=schema) == "shadow"
):
hidden_tables.append(name)
elif _is_fts_content_virtual_table(sql):
content_fts_tables.append(name)
return sorted(hidden_tables) + content_fts_tables
def _sqlite_table_type_from_schema(
conn,
table: str,
@ -150,3 +172,10 @@ def _virtual_table_module(sql: str | None) -> str | None:
if match is None:
return None
return match.group(1).strip("\"'[]`").lower()
def _is_fts_content_virtual_table(sql: str | None) -> bool:
return (
_virtual_table_module(sql) in {"fts3", "fts4", "fts5"}
and "content=" in sql.lower()
)

9
tests/test_internals_database.py
View file

@ -8,7 +8,7 @@ from datasette.app import Datasette
from datasette.database import Database, Results, MultipleValues
from datasette.database import DatasetteClosedError
from datasette.database import _deliver_write_result
from datasette.utils.sqlite import sqlite3, sqlite_version
from datasette.utils.sqlite import sqlite3
from datasette.utils import Column
import pytest
import time
@ -798,14 +798,7 @@ async def test_in_memory_databases_forbid_writes(app_client):
assert await db.table_names() == ["foo"]
def pragma_table_list_supported():
return sqlite_version()[1] >= 37
@pytest.mark.asyncio
@pytest.mark.skipif(
not pragma_table_list_supported(), reason="Requires PRAGMA table_list support"
)
async def test_hidden_tables(app_client):
ds = app_client.ds
db = ds.add_database(Database(ds, is_memory=True, is_mutable=True))

12
tests/test_utils.py
View file

@ -5,7 +5,7 @@ Tests for various datasette helper functions.
from datasette.app import Datasette
from datasette import utils
from datasette.utils.asgi import Request
from datasette.utils.sqlite import sqlite3, sqlite_table_type
from datasette.utils.sqlite import sqlite3, sqlite_hidden_table_names, sqlite_table_type
import json
import os
import pathlib
@ -246,6 +246,16 @@ def test_sqlite_table_type_detects_virtual_and_shadow_tables(monkeypatch, use_fa
assert sqlite_table_type(conn, "boxes") == "virtual"
assert sqlite_table_type(conn, "boxes_node") == "shadow"
assert sqlite_table_type(conn, "missing") is None
assert sqlite_hidden_table_names(conn) == [
"boxes_node",
"boxes_parent",
"boxes_rowid",
"search_index_config",
"search_index_content",
"search_index_data",
"search_index_docsize",
"search_index_idx",
]
finally:
conn.close()