Escape table names with [square] brackets, refs #2431 (#2846)

Several internal helpers quoted table names using SQLite [bracket]
identifiers built with an f-string, e.g. PRAGMA foreign_key_list([{table}]).
Bracket quoting cannot escape a "]" character, so any table whose name
contains "]" (for example "[foo]" or "foo]") produced
"sqlite3.OperationalError: unrecognized token" - crashing schema
introspection at startup and 500-ing the table page.

Switch these call sites to the existing escape_sqlite() helper, which uses
"double quote" quoting with correct "" escaping (the same approach already
used elsewhere in the codebase and in the test suite):

- utils/internal_db.py: PRAGMA foreign_key_list / index_list
- utils/__init__.py: get_outbound_foreign_keys
- database.py: table_counts count query
- facets.py: default "select * from" SQL

Added a regression test covering table names with "]" characters.

Co-authored-by: Claude <noreply@anthropic.com>

datasette/database.py

@@ -17,6 +17,7 @@ from .utils import (
     detect_fts,
     detect_primary_keys,
     detect_spatialite,
+    escape_sqlite,
     get_all_foreign_keys,
     get_outbound_foreign_keys,
     md5_not_usedforsecurity,
@@ -608,7 +609,7 @@ class Database:
             try:
                 table_count = (
                     await self.execute(
-                        f"select count(*) from (select * from [{table}] limit {self.count_limit + 1})",
+                        f"select count(*) from (select * from {escape_sqlite(table)} limit {self.count_limit + 1})",
                         custom_time_limit=limit,
                     )
                 ).rows[0][0]

datasette/facets.py

@@ -85,7 +85,7 @@ class Facet:
         self.database = database
         # For foreign key expansion. Can be None for e.g. stored SQL queries:
         self.table = table
-        self.sql = sql or f"select * from [{table}]"
+        self.sql = sql or f"select * from {escape_sqlite(table)}"
         self.params = params or []
         self.table_config = table_config
         # row_count can be None, in which case we calculate it ourselves:

datasette/utils/__init__.py

@@ -636,7 +636,7 @@ def detect_primary_keys(conn, table):
 
 
 def get_outbound_foreign_keys(conn, table):
-    infos = conn.execute(f"PRAGMA foreign_key_list([{table}])").fetchall()
+    infos = conn.execute(f"PRAGMA foreign_key_list({escape_sqlite(table)})").fetchall()
     fks = []
     for info in infos:
         if info is not None:

datasette/utils/internal_db.py

@@ -3,7 +3,7 @@ import textwrap
 from sqlite_utils import Database as SQLiteUtilsDatabase
 from sqlite_utils import Migrations
 
-from datasette.utils import table_column_details
+from datasette.utils import escape_sqlite, table_column_details
 
 INTERNAL_DB_SCHEMA_TABLES = {
     "catalog_databases",
@@ -213,7 +213,7 @@ async def populate_schema_tables(internal_db, db, schema_version):
                 for column in columns
             )
             foreign_keys = conn.execute(
-                f"PRAGMA foreign_key_list([{table_name}])"
+                f"PRAGMA foreign_key_list({escape_sqlite(table_name)})"
             ).fetchall()
             foreign_keys_to_insert.extend(
                 {
@@ -222,7 +222,9 @@ async def populate_schema_tables(internal_db, db, schema_version):
                 }
                 for foreign_key in foreign_keys
             )
-            indexes = conn.execute(f"PRAGMA index_list([{table_name}])").fetchall()
+            indexes = conn.execute(
+                f"PRAGMA index_list({escape_sqlite(table_name)})"
+            ).fetchall()
             indexes_to_insert.extend(
                 {
                     **{"database_name": database_name, "table_name": table_name},

tests/test_api.py

@@ -1,6 +1,6 @@
 from datasette.app import Datasette
 from datasette.plugins import DEFAULT_PLUGINS
-from datasette.utils import UNSTABLE_API_MESSAGE
+from datasette.utils import UNSTABLE_API_MESSAGE, escape_sqlite, tilde_encode
 from datasette.utils.sqlite import sqlite_version
 from datasette.version import __version__
 from .fixtures import make_app_client, EXPECTED_PLUGINS
@@ -930,6 +930,37 @@ async def test_tilde_encoded_database_names(db_name):
     assert response2.status_code == 200
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("table_name", ("[foo]", "foo]", "[foo]/bar"))
+async def test_table_with_reserved_characters_in_name(table_name):
+    # Table names containing characters such as "]" that cannot be escaped
+    # using SQLite [bracket] quoting used to break schema introspection and
+    # the table page - https://github.com/simonw/datasette/issues/2431
+    ds = Datasette()
+    db = ds.add_memory_database("test_reserved_table_names")
+    await db.execute_write(
+        "create table {} (id integer primary key, name text)".format(
+            escape_sqlite(table_name)
+        )
+    )
+    await db.execute_write(
+        "insert into {} (id, name) values (1, 'one')".format(escape_sqlite(table_name))
+    )
+    # Schema introspection (populate_schema_tables) must not crash:
+    db_response = await ds.client.get("/test_reserved_table_names.json")
+    assert db_response.status_code == 200
+    tables = {t["name"]: t for t in db_response.json()["tables"]}
+    assert tables[table_name]["count"] == 1
+    # And the table page itself must load and return the row:
+    table_response = await ds.client.get(
+        "/test_reserved_table_names/{}.json?_shape=array".format(
+            tilde_encode(table_name)
+        )
+    )
+    assert table_response.status_code == 200
+    assert table_response.json() == [{"id": 1, "name": "one"}]
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "config,expected",