rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-25 18:40:06 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

Read-only SQL query endpoint for advanced users (#820)

Browse source 
* Read-only SQL query endpoint for advanced users

POST /api/db/query runs one read-only statement; GET /api/db/schema describes
the tables.  Both are refused unless `allow_database_query: true` is set in
wrolpi.yaml (there is no UI for it).  The connection is opened with mode=ro and
an authorizer that allows only reads; queries are interrupted after 10 seconds,
rows are capped, and long values are truncated.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Database queries cannot be enabled through the API, and use bounded memory

The config update API refuses to change `allow_database_query`; only an edit of
wrolpi.yaml on the WROLPi can enable queries.  Query connections refuse values
over 32 MB and rows over 100 columns, rows are read one at a time, and the
response stops at 5 million characters.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* The database query setting can be changed like any other config

WROLPi is meant to be as secure as a bookshelf in your home; a LAN user who can
reach the API is trusted, so gating one key in the config API adds nothing.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Database queries can search FTS5 tables on SQLite 3.46

FTS5 reads `PRAGMA data_version` through the authorizer when it opens a table
on SQLite 3.46 (Debian 13), so every MATCH and the schema endpoint failed with
"vtable constructor failed".  A denied VACUUM INTO also explains itself.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* A config status test attaches the shared context it reads

It passed only when an earlier test in the same worker had attached it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
lrnselfreliance 2026-10-11 17:34:00 -06:00 • committed by GitHub
parent 3ea8eb81f6
commit f27720683d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 510 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

10
wrolpi/common.py
View file

@ -1127,6 +1127,7 @@ class MultiFileConfig:
@dataclass
class WROLPiConfigValidator:
allow_database_query: bool = None
archive_destination: str = None
check_for_upgrades: bool = None
download_daily_limit_global: int = None
@ -1373,6 +1374,7 @@ def get_config_by_file_name(file_name: str) -> ConfigFile:
class WROLPiConfig(ConfigFile):
file_name = 'wrolpi.yaml'
default_config = dict(
allow_database_query=False,
archive_destination='archive/%(domain_tag)s/%(domain)s',
download_daily_limit_global=None,
download_daily_limit_per_domain=None,
@ -1625,6 +1627,14 @@ class WROLPiConfig(ConfigFile):
def throttle_on_startup(self, value: bool):
self.update({'throttle_on_startup': value})
@property
def allow_database_query(self) -> bool:
return self._config['allow_database_query']
@allow_database_query.setter
def allow_database_query(self, value: bool):
self.update({'allow_database_query': value})
@property
def wrol_mode(self) -> bool:
return self._config['wrol_mode']

251
wrolpi/db_query.py Normal file
View file

@ -0,0 +1,251 @@
"""Read-only SQL access to the WROLPi database, for advanced users and agents (the WROLPi MCP).
Disabled unless `allow_database_query: true` is set in wrolpi.yaml; there is no UI for this setting.
A query can never change the database:
* The connection is opened with `mode=ro`, so SQLite itself refuses every write.
* An authorizer permits only reads (SELECT, CTEs, functions, and introspection PRAGMAs), so a statement
cannot ATTACH another file, create TEMP objects, or change a PRAGMA.
A query cannot monopolize the device:
* A progress handler interrupts it after `QUERY_TIMEOUT` seconds (this also bounds how long it pins a
WAL snapshot).
* SQLite refuses to build a value longer than `MAX_VALUE_LENGTH`, or a row wider than `MAX_COLUMNS`.
* Rows are read one at a time; long values are cut to `MAX_CELL_LENGTH`, and rows stop at `MAX_LIMIT` or once
the response holds `MAX_RESPONSE_LENGTH` characters.
"""
import asyncio
import contextlib
import sqlite3
import time
from dataclasses import dataclass, field
from http import HTTPStatus
from typing import List, Optional
from sanic import Blueprint
from sanic.request import Request
from sanic_ext import validate
from sanic_ext.extensions.openapi import openapi
from wrolpi.api_utils import json_response
from wrolpi.common import logger, get_wrolpi_config
from wrolpi.db import get_db_file
from wrolpi.errors import DatabaseQueryDisabled, InvalidDatabaseQuery
from wrolpi.schema import JSONErrorResponse
logger = logger.getChild(__name__)
DEFAULT_LIMIT = 500
MAX_LIMIT = 5000
MAX_CELL_LENGTH = 10_000
MAX_RESPONSE_LENGTH = 5_000_000
# Larger than any value WROLPi stores (a document's text), far smaller than SQLite's 1 GB default.
MAX_VALUE_LENGTH = 32 * 1024 * 1024
MAX_COLUMNS = 100
QUERY_TIMEOUT = 10.0
# How many SQLite VM instructions run between timeout checks.
PROGRESS_INTERVAL = 10_000
ALLOWED_ACTIONS = {
sqlite3.SQLITE_SELECT,
sqlite3.SQLITE_READ,
sqlite3.SQLITE_FUNCTION,
sqlite3.SQLITE_RECURSIVE,
}
# PRAGMAs which only read, whatever their argument. Every other PRAGMA (e.g. `PRAGMA foreign_keys=OFF`) is refused.
ALLOWED_PRAGMAS = {
# FTS5 reads this when it opens a table, through the authorizer, on SQLite 3.46 (Debian 13).
'data_version',
'foreign_key_list',
'index_info',
'index_list',
'index_xinfo',
'table_info',
'table_list',
'table_xinfo',
}
DENIED_FUNCTIONS = {'load_extension'}
def _authorizer(action: int, arg1: Optional[str], arg2: Optional[str], _db: Optional[str], _trigger: Optional[str]):
if action == sqlite3.SQLITE_PRAGMA:
if arg1 and arg1.lower() in ALLOWED_PRAGMAS:
return sqlite3.SQLITE_OK
return sqlite3.SQLITE_DENY
if action == sqlite3.SQLITE_FUNCTION and arg2 and arg2.lower() in DENIED_FUNCTIONS:
return sqlite3.SQLITE_DENY
if action in ALLOWED_ACTIONS:
return sqlite3.SQLITE_OK
return sqlite3.SQLITE_DENY
@contextlib.contextmanager
def read_only_connection(timeout: float = QUERY_TIMEOUT):
"""A connection to the WROLPi database which can only read, and which interrupts any statement still running
`timeout` seconds after the connection was opened."""
conn = sqlite3.connect(f'file:{get_db_file()}?mode=ro', uri=True, timeout=5, check_same_thread=False)
try:
deadline = time.monotonic() + timeout
conn.set_progress_handler(lambda: int(time.monotonic() > deadline), PROGRESS_INTERVAL)
conn.set_authorizer(_authorizer)
conn.setlimit(sqlite3.SQLITE_LIMIT_LENGTH, MAX_VALUE_LENGTH)
conn.setlimit(sqlite3.SQLITE_LIMIT_COLUMN, MAX_COLUMNS)
yield conn
finally:
conn.close()
def _cell(value):
if isinstance(value, bytes):
return f'<blob {len(value)} bytes>'
if isinstance(value, str) and len(value) > MAX_CELL_LENGTH:
return value[:MAX_CELL_LENGTH] + f'... <truncated {len(value) - MAX_CELL_LENGTH} characters>'
return value
def _cell_length(value) -> int:
return len(value) if isinstance(value, str) else 8
def _raise_query_error(e: sqlite3.Error):
message = str(e)
if isinstance(e, sqlite3.OperationalError) and message == 'interrupted':
message = f'Query exceeded the {QUERY_TIMEOUT:g} second time limit'
elif 'not authorized' in message or 'authorization denied' in message:
message = 'Only read-only queries (SELECT) are allowed'
elif 'too big' in message:
message = f'A value is too large (over {MAX_VALUE_LENGTH} bytes)'
raise InvalidDatabaseQuery(message) from e
def run_query(sql: str, limit: int = DEFAULT_LIMIT) -> dict:
"""Run one read-only statement, return at most `limit` rows."""
if not sql or not sql.strip():
raise InvalidDatabaseQuery('Query is empty')
limit = max(1, min(int(limit or DEFAULT_LIMIT), MAX_LIMIT))
start = time.monotonic()
rows, truncated, response_length = [], False, 0
with read_only_connection() as conn:
try:
curs = conn.execute(sql)
columns = [i[0] for i in curs.description or []]
# One row at a time, so only the kept (truncated) values are held.
for row in curs:
if len(rows) >= limit or response_length >= MAX_RESPONSE_LENGTH:
truncated = True
break
row = [_cell(i) for i in row]
response_length += sum(_cell_length(i) for i in row)
rows.append(row)
except sqlite3.Error as e:
_raise_query_error(e)
except sqlite3.Warning as e:
# Python refuses more than one statement.
raise InvalidDatabaseQuery(str(e)) from e
return dict(
columns=columns,
rows=rows,
row_count=len(rows),
truncated=truncated,
elapsed=round(time.monotonic() - start, 3),
)
def _is_fts_shadow_table(name: str, virtual_tables: List[str]) -> bool:
return any(name.startswith(f'{i}_') for i in virtual_tables)
def get_schema() -> List[dict]:
"""Describe every table and view (but not SQLite internals or FTS5's shadow tables)."""
with read_only_connection() as conn:
objects = conn.execute(
"SELECT name, type, sql FROM sqlite_master"
" WHERE type IN ('table', 'view') AND name NOT LIKE 'sqlite_%' ORDER BY name").fetchall()
virtual_tables = [name for name, _, sql in objects if sql and sql.upper().startswith('CREATE VIRTUAL TABLE')]
tables = []
for name, type_, sql in objects:
if _is_fts_shadow_table(name, virtual_tables):
continue
is_virtual = name in virtual_tables
columns = [
dict(name=col_name, type=col_type, not_null=bool(not_null), primary_key=bool(pk))
for _, col_name, col_type, not_null, _, pk in conn.execute(
'SELECT * FROM pragma_table_info(?)', (name,)).fetchall()
]
indexes = []
if type_ == 'table' and not is_virtual:
for _, index_name, unique, *_ in conn.execute(
'SELECT * FROM pragma_index_list(?)', (name,)).fetchall():
index_columns = [i[2] for i in conn.execute(
'SELECT * FROM pragma_index_info(?)', (index_name,)).fetchall()]
indexes.append(dict(name=index_name, unique=bool(unique), columns=index_columns))
table = dict(
name=name,
type='virtual' if is_virtual else type_,
columns=columns,
indexes=indexes,
)
if is_virtual or type_ == 'view':
# The definition explains an FTS5 table's tokenizer and content table, or a view's query.
table['sql'] = sql
tables.append(table)
return tables
def check_database_query_allowed():
if not get_wrolpi_config().allow_database_query:
raise DatabaseQueryDisabled()
db_query_bp = Blueprint('DatabaseQuery', url_prefix='/api/db')
@dataclass
class DatabaseQueryRequest:
sql: str
limit: Optional[int] = DEFAULT_LIMIT
@dataclass
class DatabaseQueryResponse:
columns: List[str]
rows: List[list]
row_count: int
truncated: bool
elapsed: float
@dataclass
class DatabaseSchemaResponse:
tables: List[dict] = field(default_factory=list)
@db_query_bp.post('/query')
@openapi.definition(
description='Run one read-only SQL statement (SELECT) against the WROLPi database. Requires'
' `allow_database_query: true` in wrolpi.yaml.',
body=DatabaseQueryRequest,
)
@openapi.response(HTTPStatus.OK, DatabaseQueryResponse)
@openapi.response(HTTPStatus.BAD_REQUEST, JSONErrorResponse)
@openapi.response(HTTPStatus.FORBIDDEN, JSONErrorResponse)
@validate(DatabaseQueryRequest)
async def post_query(_: Request, body: DatabaseQueryRequest):
check_database_query_allowed()
result = await asyncio.to_thread(run_query, body.sql, body.limit)
return json_response(result)
@db_query_bp.get('/schema')
@openapi.definition(
description='Describe the tables, columns, and indexes of the WROLPi database. Requires'
' `allow_database_query: true` in wrolpi.yaml.',
)
@openapi.response(HTTPStatus.OK, DatabaseSchemaResponse)
@openapi.response(HTTPStatus.FORBIDDEN, JSONErrorResponse)
async def get_schema_request(_: Request):
check_database_query_allowed()
tables = await asyncio.to_thread(get_schema)
return json_response(dict(tables=tables))

12
wrolpi/errors.py
View file

@ -229,3 +229,15 @@ class InvalidJob(APIError):
code = 'INVALID_JOB'
summary = 'The Job request is invalid.'
status_code = HTTPStatus.BAD_REQUEST
class DatabaseQueryDisabled(APIError):
code = 'DATABASE_QUERY_DISABLED'
summary = 'Database queries are disabled. Set `allow_database_query: true` in wrolpi.yaml, then import it.'
status_code = HTTPStatus.FORBIDDEN
class InvalidDatabaseQuery(APIError):
code = 'INVALID_DATABASE_QUERY'
summary = 'The database query is invalid, not read-only, or took too long.'
status_code = HTTPStatus.BAD_REQUEST

2
wrolpi/root_api.py
View file

@ -30,6 +30,7 @@ from wrolpi.common import logger, get_wrolpi_config, wrol_mode_enabled, get_medi
wrol_mode_check, native_only, disable_wrol_mode, enable_wrol_mode, get_global_statistics, url_strip_host, \
set_global_log_level, get_relative_to_media_directory, search_other_estimates, set_system_timezone
from wrolpi.config_api import config_bp
from wrolpi.db_query import db_query_bp
from wrolpi.dates import now
from wrolpi.downloader import download_manager
from wrolpi.errors import WROLModeEnabled, InvalidConfig, ValidationError
@ -59,6 +60,7 @@ api_app.blueprint(archive_bp)
api_app.blueprint(bookmarks_bp)
api_app.blueprint(collection_bp) # Unified collection endpoints
api_app.blueprint(config_bp)
api_app.blueprint(db_query_bp)
api_app.blueprint(docs_bp)
api_app.blueprint(files_bp)
api_app.blueprint(flasher_bp)

4
wrolpi/test/test_config_backup_import.py
View file

@ -549,7 +549,9 @@ class TestConfigStatusIncludesBackupInfo:
assert status['has_backup_import'] is True
assert 'backup_dates' not in status
def test_non_db_config_no_backup_info(self, test_directory, test_wrolpi_config):
@pytest.mark.asyncio
async def test_non_db_config_no_backup_info(self, test_directory, test_wrolpi_config, async_client):
"""`config_status` reads the shared context, which `async_client` attaches."""
from wrolpi.common import get_wrolpi_config
config = get_wrolpi_config()
status = config.config_status()

232
wrolpi/test/test_db_query.py Normal file
View file

@ -0,0 +1,232 @@
import json
import sqlite3
from http import HTTPStatus
from unittest import mock
import pytest
from wrolpi import db_query
from wrolpi.common import get_wrolpi_config
from wrolpi.db import get_db_file
@pytest.fixture
def allow_database_query(test_wrolpi_config):
get_wrolpi_config().allow_database_query = True
yield
get_wrolpi_config().allow_database_query = False
async def post_query(async_client, sql: str, **kwargs):
body = dict(sql=sql, **kwargs)
return await async_client.post('/api/db/query', content=json.dumps(body))
@pytest.mark.asyncio
async def test_database_query_disabled_by_default(test_session, async_client, test_wrolpi_config):
"""Both endpoints refuse until the setting is enabled in wrolpi.yaml."""
assert get_wrolpi_config().allow_database_query is False
request, response = await post_query(async_client, 'SELECT 1')
assert response.status_code == HTTPStatus.FORBIDDEN
assert response.json['code'] == 'DATABASE_QUERY_DISABLED'
request, response = await async_client.get('/api/db/schema')
assert response.status_code == HTTPStatus.FORBIDDEN
assert response.json['code'] == 'DATABASE_QUERY_DISABLED'
@pytest.mark.asyncio
async def test_database_query_enabled_by_config_import(test_session, async_client, test_wrolpi_config):
"""An advanced user enables queries by editing wrolpi.yaml and importing it; there is no UI for it."""
config = get_wrolpi_config()
config.dump_config()
contents = test_wrolpi_config.read_text()
assert 'allow_database_query: false' in contents
test_wrolpi_config.write_text(contents.replace('allow_database_query: false', 'allow_database_query: true'))
request, response = await async_client.post('/api/config/import',
content=json.dumps(dict(file_name='wrolpi.yaml')))
assert response.status_code == HTTPStatus.NO_CONTENT
try:
request, response = await post_query(async_client, 'SELECT 1 AS one')
assert response.status_code == HTTPStatus.OK, response.json
assert response.json['rows'] == [[1]]
finally:
config.allow_database_query = False
@pytest.mark.asyncio
async def test_database_query_select(test_session, async_client, allow_database_query, tag_factory):
"""A SELECT returns committed rows, with column names."""
await tag_factory('one')
await tag_factory('two')
test_session.commit()
request, response = await post_query(async_client, 'SELECT name FROM tag ORDER BY name')
assert response.status_code == HTTPStatus.OK, response.json
assert response.json['columns'] == ['name']
assert response.json['rows'] == [['one'], ['two']]
assert response.json['row_count'] == 2
assert response.json['truncated'] is False
# CTEs (including recursive ones) are reads.
sql = 'WITH RECURSIVE n(x) AS (SELECT 1 UNION ALL SELECT x + 1 FROM n WHERE x < 3) SELECT x FROM n'
request, response = await post_query(async_client, sql)
assert response.status_code == HTTPStatus.OK, response.json
assert response.json['rows'] == [[1], [2], [3]]
@pytest.mark.asyncio
@pytest.mark.parametrize('sql', [
"INSERT INTO tag (name, color) VALUES ('evil', '#000000')",
"UPDATE tag SET name = 'evil'",
'DELETE FROM tag',
'DROP TABLE tag',
'CREATE TABLE evil (id INTEGER)',
'CREATE TEMP TABLE evil (id INTEGER)',
"ATTACH DATABASE ':memory:' AS evil",
'PRAGMA journal_mode=DELETE',
'PRAGMA foreign_keys=OFF',
'BEGIN',
'VACUUM',
"VACUUM INTO 'copy.db'",
"SELECT load_extension('evil')",
"SELECT 1; DELETE FROM tag",
])
async def test_database_query_refuses_writes(test_session, async_client, allow_database_query, tag_factory, sql):
"""Anything other than a read is refused, and the database is unchanged."""
await tag_factory('one')
test_session.commit()
request, response = await post_query(async_client, sql)
assert response.status_code == HTTPStatus.BAD_REQUEST, response.json
assert response.json['code'] == 'INVALID_DATABASE_QUERY'
if ';' not in sql:
assert response.json['error'] == 'Only read-only queries (SELECT) are allowed'
request, response = await post_query(async_client, 'SELECT name FROM tag')
assert response.json['rows'] == [['one']]
@pytest.mark.asyncio
async def test_database_query_connection_is_read_only(test_session, allow_database_query):
"""Even without the authorizer, SQLite refuses to write through the connection."""
with db_query.read_only_connection() as conn:
conn.set_authorizer(None)
with pytest.raises(Exception, match='readonly'):
conn.execute("INSERT INTO tag (name, color) VALUES ('evil', '#000000')")
@pytest.mark.asyncio
async def test_database_query_introspection_pragmas(test_session, async_client, allow_database_query):
"""Schema-describing PRAGMAs are allowed, both as statements and as table-valued functions."""
request, response = await post_query(async_client, 'PRAGMA table_info(tag)')
assert response.status_code == HTTPStatus.OK, response.json
assert 'name' in [row[1] for row in response.json['rows']]
request, response = await post_query(async_client, "SELECT name FROM pragma_table_info('tag')")
assert response.status_code == HTTPStatus.OK, response.json
assert ['name'] in response.json['rows']
@pytest.mark.asyncio
async def test_database_query_errors(test_session, async_client, allow_database_query):
request, response = await post_query(async_client, 'SELECT * FROM no_such_table')
assert response.status_code == HTTPStatus.BAD_REQUEST
assert 'no such table' in response.json['error']
request, response = await post_query(async_client, ' ')
assert response.status_code == HTTPStatus.BAD_REQUEST
assert response.json['code'] == 'INVALID_DATABASE_QUERY'
@pytest.mark.asyncio
async def test_database_query_limits(test_session, async_client, allow_database_query):
"""Rows are capped by `limit` (and MAX_LIMIT), long text is truncated, and blobs are summarized."""
sql = 'WITH RECURSIVE n(x) AS (SELECT 1 UNION ALL SELECT x + 1 FROM n WHERE x < 20) SELECT x FROM n'
request, response = await post_query(async_client, sql, limit=5)
assert response.status_code == HTTPStatus.OK, response.json
assert response.json['rows'] == [[1], [2], [3], [4], [5]]
assert response.json['truncated'] is True
with mock.patch.object(db_query, 'MAX_LIMIT', 3):
request, response = await post_query(async_client, sql, limit=1000)
assert response.json['row_count'] == 3
assert response.json['truncated'] is True
request, response = await post_query(
async_client, f"SELECT printf('%.*c', {db_query.MAX_CELL_LENGTH + 5}, 'a'), zeroblob(16)")
text, blob = response.json['rows'][0]
assert text.startswith('a' * db_query.MAX_CELL_LENGTH + '...')
assert text.endswith('<truncated 5 characters>')
assert blob == '<blob 16 bytes>'
@pytest.mark.asyncio
async def test_database_query_timeout(test_session, async_client, allow_database_query):
"""A query that runs too long is interrupted."""
sql = 'WITH RECURSIVE n(x) AS (SELECT 1 UNION ALL SELECT x + 1 FROM n) SELECT count(*) FROM n'
with mock.patch.object(db_query, 'QUERY_TIMEOUT', 0.2):
request, response = await post_query(async_client, sql)
assert response.status_code == HTTPStatus.BAD_REQUEST
assert 'time limit' in response.json['error']
@pytest.mark.asyncio
async def test_database_schema(test_session, async_client, allow_database_query):
request, response = await async_client.get('/api/db/schema')
assert response.status_code == HTTPStatus.OK, response.json
tables = {i['name']: i for i in response.json['tables']}
tag = tables['tag']
assert tag['type'] == 'table'
columns = {i['name']: i for i in tag['columns']}
assert columns['id']['primary_key'] is True
assert 'name' in columns
assert any(i['columns'] == ['name'] for i in tag['indexes'])
# FTS5 tables are described (with their definition), but their shadow tables are not.
virtual = [i for i in tables.values() if i['type'] == 'virtual']
assert virtual and all('fts5' in i['sql'].lower() for i in virtual)
for table in virtual:
assert f'{table["name"]}_data' not in tables
assert not any(name.startswith('sqlite_') for name in tables)
assert get_db_file().is_file()
@pytest.mark.asyncio
async def test_database_query_memory_limits(test_session, async_client, allow_database_query):
"""A query cannot build huge values or rows, and the response is bounded however many rows are asked for."""
request, response = await post_query(async_client, f'SELECT zeroblob({db_query.MAX_VALUE_LENGTH + 1})')
assert response.status_code == HTTPStatus.BAD_REQUEST
assert 'too large' in response.json['error']
columns = ', '.join(['1'] * (db_query.MAX_COLUMNS + 1))
request, response = await post_query(async_client, f'SELECT {columns}')
assert response.status_code == HTTPStatus.BAD_REQUEST
# Every row is a full-size cell; rows stop once the response budget is spent.
sql = (f"WITH RECURSIVE n(x) AS (SELECT 1 UNION ALL SELECT x + 1 FROM n WHERE x < 100)"
f" SELECT printf('%.*c', {db_query.MAX_CELL_LENGTH}, 'a') FROM n")
with mock.patch.object(db_query, 'MAX_RESPONSE_LENGTH', db_query.MAX_CELL_LENGTH * 3):
request, response = await post_query(async_client, sql, limit=100)
assert response.status_code == HTTPStatus.OK, response.json
assert response.json['row_count'] == 3
assert response.json['truncated'] is True
@pytest.mark.asyncio
async def test_database_query_full_text_search(test_session, async_client, allow_database_query, video_factory):
"""FTS5 tables can be searched. FTS5 reads `PRAGMA data_version` through the authorizer on some SQLite
versions (3.46), so that PRAGMA must be allowed even where this SQLite does not ask."""
video_factory(title='Catching rain water')
test_session.commit()
sql = "SELECT fg.title FROM file_group_fts JOIN file_group fg ON fg.id = file_group_fts.rowid" \
" WHERE file_group_fts MATCH 'water'"
request, response = await post_query(async_client, sql)
assert response.status_code == HTTPStatus.OK, response.json
assert response.json['rows'] == [['Catching rain water']]
assert db_query._authorizer(sqlite3.SQLITE_PRAGMA, 'data_version', None, 'main', None) == sqlite3.SQLITE_OK