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

Black formatting

Browse source
This commit is contained in:
Simon Willison 2026-02-17 13:30:24 -08:00
commit 5c3137d148
31 changed files with 82 additions and 222 deletions
Download patch file Download diff file Expand all files Collapse all files

10
datasette/app.py
View file

@ -633,9 +633,7 @@ class Datasette:
"""
INSERT OR REPLACE INTO catalog_databases (database_name, path, is_memory, schema_version)
VALUES {}
""".format(
placeholders
),
""".format(placeholders),
values,
)
await populate_schema_tables(internal_db, db)
@ -813,14 +811,12 @@ class Datasette:
return orig
async def get_instance_metadata(self):
rows = await self.get_internal_database().execute(
"""
rows = await self.get_internal_database().execute("""
SELECT
key,
value
FROM metadata_instance
"""
)
""")
return dict(rows)
async def get_database_metadata(self, database_name: str):

8
datasette/cli.py
View file

@ -109,15 +109,11 @@ def sqlite_extensions(fn):
return fn(*args, **kwargs)
except AttributeError as e:
if "enable_load_extension" in str(e):
raise click.ClickException(
textwrap.dedent(
"""
raise click.ClickException(textwrap.dedent("""
Your Python installation does not have the ability to load SQLite extensions.
More information: https://datasette.io/help/extensions
"""
).strip()
)
""").strip())
raise
return wrapped

36
datasette/database.py
View file

@ -532,10 +532,7 @@ class Database:
]
if sqlite_version()[1] >= 37:
hidden_tables += [
x[0]
for x in await self.execute(
"""
hidden_tables += [x[0] for x in await self.execute("""
with shadow_tables as (
select name
from pragma_table_list
@ -554,14 +551,9 @@ class Database:
select name from core_tables
)
select name from combined order by 1
"""
)
]
""")]
else:
hidden_tables += [
x[0]
for x in await self.execute(
"""
hidden_tables += [x[0] for x in await self.execute("""
WITH base AS (
SELECT name
FROM sqlite_master
@ -607,22 +599,15 @@ class Database:
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(
"""
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=%'
"""
)
]
""")]
has_spatialite = await self.execute_fn(detect_spatialite)
if has_spatialite:
@ -641,16 +626,11 @@ class Database:
"KNN",
"KNN2",
] + [
r[0]
for r in (
await self.execute(
"""
r[0] for r in (await self.execute("""
select name from sqlite_master
where name like "idx_%"
and type = "table"
"""
)
).rows
""")).rows
]
return hidden_tables

1
datasette/default_permissions/defaults.py
View file

@ -14,7 +14,6 @@ if TYPE_CHECKING:
from datasette import hookimpl
from datasette.permissions import PermissionSQL
# Actions that are allowed by default (unless --default-deny is used)
DEFAULT_ALLOW_ACTIONS = frozenset(
{

12
datasette/facets.py
View file

@ -233,9 +233,7 @@ class ColumnFacet(Facet):
)
where {col} is not null
group by {col} order by count desc, value limit {limit}
""".format(
col=escape_sqlite(column), sql=self.sql, limit=facet_size + 1
)
""".format(col=escape_sqlite(column), sql=self.sql, limit=facet_size + 1)
try:
facet_rows_results = await self.ds.execute(
self.database,
@ -482,9 +480,7 @@ class DateFacet(Facet):
select date({column}) from (
select * from ({sql}) limit 100
) where {column} glob "????-??-*"
""".format(
column=escape_sqlite(column), sql=self.sql
)
""".format(column=escape_sqlite(column), sql=self.sql)
try:
results = await self.ds.execute(
self.database,
@ -530,9 +526,7 @@ class DateFacet(Facet):
)
where date({col}) is not null
group by date({col}) order by count desc, value limit {limit}
""".format(
col=escape_sqlite(column), sql=self.sql, limit=facet_size + 1
)
""".format(col=escape_sqlite(column), sql=self.sql, limit=facet_size + 1)
try:
facet_rows_results = await self.ds.execute(
self.database,

17
datasette/inspect.py
View file

@ -10,7 +10,6 @@ from .utils import (
sqlite3,
)
HASH_BLOCK_SIZE = 1024 * 1024
@ -70,16 +69,11 @@ def inspect_tables(conn, database_metadata):
tables[table]["foreign_keys"] = info
# Mark tables 'hidden' if they relate to FTS virtual tables
hidden_tables = [
r["name"]
for r in conn.execute(
"""
hidden_tables = [r["name"] for r in conn.execute("""
select name from sqlite_master
where rootpage = 0
and sql like '%VIRTUAL TABLE%USING FTS%'
"""
)
]
""")]
if detect_spatialite(conn):
# Also hide Spatialite internal tables
@ -94,14 +88,11 @@ def inspect_tables(conn, database_metadata):
"views_geometry_columns",
"virts_geometry_columns",
] + [
r["name"]
for r in conn.execute(
"""
r["name"] for r in conn.execute("""
select name from sqlite_master
where name like "idx_%"
and type = "table"
"""
)
""")
]
for t in tables.keys():

1
datasette/permissions.py
View file

@ -3,7 +3,6 @@ from dataclasses import dataclass
from typing import Any, NamedTuple
import contextvars
# Context variable to track when permission checks should be skipped
_skip_permission_checks = contextvars.ContextVar(
"skip_permission_checks", default=False

4
datasette/utils/__init__.py
View file

@ -677,9 +677,7 @@ def detect_fts_sql(table):
and sql like '%VIRTUAL TABLE%USING FTS%'
)
)
""".format(
table=table.replace("'", "''")
)
""".format(table=table.replace("'", "''"))
def detect_json1(conn=None):

18
datasette/utils/actions_sql.py
View file

@ -180,13 +180,11 @@ async def _build_single_action_sql(
# Skip plugins that only provide restriction_sql (no permission rules)
if permission_sql.sql is None:
continue
rule_sqls.append(
f"""
rule_sqls.append(f"""
SELECT parent, child, allow, reason, '{permission_sql.source}' AS source_plugin FROM (
{permission_sql.sql}
)
""".strip()
)
""".strip())
# If no rules, return empty result (deny all)
if not rule_sqls:
@ -405,14 +403,12 @@ async def _build_single_action_sql(
# Add restriction filter if there are restrictions
if restriction_sqls:
query_parts.append(
"""
query_parts.append("""
AND EXISTS (
SELECT 1 FROM restriction_list r
WHERE (r.parent = decisions.parent OR r.parent IS NULL)
AND (r.child = decisions.child OR r.child IS NULL)
)"""
)
)""")
# Add parent filter if specified
if parent is not None:
@ -479,13 +475,11 @@ async def build_permission_rules_sql(
if permission_sql.sql is None:
continue
union_parts.append(
f"""
union_parts.append(f"""
SELECT parent, child, allow, reason, '{permission_sql.source}' AS source_plugin FROM (
{permission_sql.sql}
)
""".strip()
)
""".strip())
rules_union = " UNION ALL ".join(union_parts)
return rules_union, all_params, restriction_sqls

14
datasette/utils/internal_db.py
View file

@ -3,8 +3,7 @@ from datasette.utils import table_column_details
async def init_internal_db(db):
create_tables_sql = textwrap.dedent(
"""
create_tables_sql = textwrap.dedent("""
CREATE TABLE IF NOT EXISTS catalog_databases (
database_name TEXT PRIMARY KEY,
path TEXT,
@ -68,16 +67,13 @@ async def init_internal_db(db):
FOREIGN KEY (database_name) REFERENCES catalog_databases(database_name),
FOREIGN KEY (database_name, table_name) REFERENCES catalog_tables(database_name, table_name)
);
"""
).strip()
""").strip()
await db.execute_write_script(create_tables_sql)
await initialize_metadata_tables(db)
async def initialize_metadata_tables(db):
await db.execute_write_script(
textwrap.dedent(
"""
await db.execute_write_script(textwrap.dedent("""
CREATE TABLE IF NOT EXISTS metadata_instance (
key text,
value text,
@ -107,9 +103,7 @@ async def initialize_metadata_tables(db):
value text,
unique(database_name, resource_name, column_name, key)
);
"""
)
)
"""))
async def populate_schema_tables(internal_db, db):

7
datasette/utils/permissions.py
View file

@ -9,7 +9,6 @@ from datasette.permissions import PermissionSQL
from datasette.plugins import pm
from datasette.utils import await_me_maybe
# Sentinel object to indicate permission checks should be skipped
SKIP_PERMISSION_CHECKS = object()
@ -116,13 +115,11 @@ def build_rules_union(
if p.sql is None:
continue
parts.append(
f"""
parts.append(f"""
SELECT parent, child, allow, reason, '{p.source}' AS source_plugin FROM (
{p.sql}
)
""".strip()
)
""".strip())
if not parts:
# Empty UNION that returns no rows

8
datasette/views/base.py
View file

@ -241,8 +241,7 @@ class DataView(BaseView):
data, extra_template_data, templates = response_or_template_contexts
except QueryInterrupted as ex:
raise DatasetteError(
textwrap.dedent(
"""
textwrap.dedent("""
<p>SQL query took too long. The time limit is controlled by the
<a href="https://docs.datasette.io/en/stable/settings.html#sql-time-limit-ms">sql_time_limit_ms</a>
configuration option.</p>
@ -251,10 +250,7 @@ class DataView(BaseView):
let ta = document.querySelector("textarea");
ta.style.height = ta.scrollHeight + "px";
</script>
""".format(
escape(ex.sql)
)
).strip(),
""".format(escape(ex.sql))).strip(),
title="SQL Interrupted",
status=400,
message_is_html=True,

8
datasette/views/database.py
View file

@ -615,8 +615,7 @@ class QueryView(View):
rows = results.rows
except QueryInterrupted as ex:
raise DatasetteError(
textwrap.dedent(
"""
textwrap.dedent("""
<p>SQL query took too long. The time limit is controlled by the
<a href="https://docs.datasette.io/en/stable/settings.html#sql-time-limit-ms">sql_time_limit_ms</a>
configuration option.</p>
@ -625,10 +624,7 @@ class QueryView(View):
let ta = document.querySelector("textarea");
ta.style.height = ta.scrollHeight + "px";
</script>
""".format(
markupsafe.escape(ex.sql)
)
).strip(),
""".format(markupsafe.escape(ex.sql))).strip(),
title="SQL Interrupted",
status=400,
message_is_html=True,

1
datasette/views/index.py
View file

@ -12,7 +12,6 @@ from datasette.version import __version__
from .base import BaseView
# Truncate table list on homepage at:
TRUNCATE_AT = 5

1
datasette/views/special.py
View file

@ -13,7 +13,6 @@ from .base import BaseView, View
import secrets
import urllib
logger = logging.getLogger(__name__)

1
tests/conftest.py
View file

@ -11,7 +11,6 @@ import time
from dataclasses import dataclass
from datasette import Event, hookimpl
try:
import pysqlite3 as sqlite3
except ImportError:

20
tests/fixtures.py
View file

@ -13,7 +13,6 @@ import string
import tempfile
import textwrap
# This temp file is used by one of the plugin config tests
TEMP_PLUGIN_SECRET_FILE = os.path.join(tempfile.gettempdir(), "plugin-secret")
@ -331,16 +330,14 @@ CONFIG = {
"sql": "select :_header_user_agent as user_agent, :_now_datetime_utc as datetime",
},
"neighborhood_search": {
"sql": textwrap.dedent(
"""
"sql": textwrap.dedent("""
select _neighborhood, facet_cities.name, state
from facetable
join facet_cities
on facetable._city_id = facet_cities.id
where _neighborhood like '%' || :text || '%'
order by _neighborhood;
"""
),
"""),
"title": "Search neighborhoods",
"description_html": "<b>Demonstrating</b> simple like search",
"fragment": "fragment-goes-here",
@ -710,19 +707,10 @@ CREATE VIEW searchable_view_configured_by_metadata AS
for a, b, c, content in generate_compound_rows(1001)
]
)
+ "\n".join(
[
"""INSERT INTO sortable VALUES (
+ "\n".join(["""INSERT INTO sortable VALUES (
"{pk1}", "{pk2}", "{content}", {sortable},
{sortable_with_nulls}, {sortable_with_nulls_2}, "{text}");
""".format(
**row
).replace(
"None", "null"
)
for row in generate_sortable_rows(201)
]
)
""".format(**row).replace("None", "null") for row in generate_sortable_rows(201)])
)
TABLE_PARAMETERIZED_SQL = [
("insert into binary_data (data) values (?);", [b"\x15\x1c\x02\xc7\xad\x05\xfe"]),

8
tests/plugins/my_plugin.py
View file

@ -261,8 +261,7 @@ def register_routes():
response = Response.redirect("/")
datasette.set_actor_cookie(response, {"id": "root"})
return response
return Response.html(
"""
return Response.html("""
<form action="{}" method="POST">
<p>
<input type="hidden" name="csrftoken" value="{}">
@ -271,10 +270,7 @@ def register_routes():
style="font-size: 2em; padding: 0.1em 0.5em;">
</p>
</form>
""".format(
request.path, request.scope["csrftoken"]()
)
)
""".format(request.path, request.scope["csrftoken"]()))
def asgi_scope(scope):
return Response.json(scope, default=repr)

8
tests/test_cli.py
View file

@ -115,13 +115,9 @@ def test_plugins_cli(app_client):
def test_metadata_yaml():
yaml_file = io.StringIO(
textwrap.dedent(
"""
yaml_file = io.StringIO(textwrap.dedent("""
title: Hello from YAML
"""
)
)
"""))
# Annoyingly we have to provide all default arguments here:
ds = serve.callback(
[],

4
tests/test_cli_serve_get.py
View file

@ -16,9 +16,7 @@ def test_serve_with_get(tmp_path_factory):
def startup(datasette):
with open("{}", "w") as fp:
fp.write("hello")
""".format(
str(plugins_dir / "hello.txt")
),
""".format(str(plugins_dir / "hello.txt")),
),
"utf-8",
)

6
tests/test_config_dir.py
View file

@ -51,8 +51,7 @@ def config_dir(tmp_path_factory):
for dbname in ("demo.db", "immutable.db", "j.sqlite3", "k.sqlite"):
db = sqlite3.connect(str(config_dir / dbname))
db.executescript(
"""
db.executescript("""
CREATE TABLE cities (
id integer primary key,
name text
@ -60,8 +59,7 @@ def config_dir(tmp_path_factory):
INSERT INTO cities (id, name) VALUES
(1, 'San Francisco')
;
"""
)
""")
# Mark "immutable.db" as immutable
(config_dir / "inspect-data.json").write_text(

22
tests/test_csv.py
View file

@ -9,16 +9,12 @@ EXPECTED_TABLE_CSV = """id,content
3,
4,RENDER_CELL_DEMO
5,RENDER_CELL_ASYNC
""".replace(
"\n", "\r\n"
)
""".replace("\n", "\r\n")
EXPECTED_CUSTOM_CSV = """content
hello
world
""".replace(
"\n", "\r\n"
)
""".replace("\n", "\r\n")
EXPECTED_TABLE_WITH_LABELS_CSV = """
pk,created,planet_int,on_earth,state,_city_id,_city_id_label,_neighborhood,tags,complex_array,distinct_some_null,n
@ -37,17 +33,13 @@ pk,created,planet_int,on_earth,state,_city_id,_city_id_label,_neighborhood,tags,
13,2019-01-17 08:00:00,1,1,MI,3,Detroit,Corktown,[],[],,
14,2019-01-17 08:00:00,1,1,MI,3,Detroit,Mexicantown,[],[],,
15,2019-01-17 08:00:00,2,0,MC,4,Memnonia,Arcadia Planitia,[],[],,
""".lstrip().replace(
"\n", "\r\n"
)
""".lstrip().replace("\n", "\r\n")
EXPECTED_TABLE_WITH_NULLABLE_LABELS_CSV = """
pk,foreign_key_with_label,foreign_key_with_label_label,foreign_key_with_blank_label,foreign_key_with_blank_label_label,foreign_key_with_no_label,foreign_key_with_no_label_label,foreign_key_compound_pk1,foreign_key_compound_pk2
1,1,hello,3,,1,1,a,b
2,,,,,,,,
""".lstrip().replace(
"\n", "\r\n"
)
""".lstrip().replace("\n", "\r\n")
@pytest.mark.asyncio
@ -108,8 +100,7 @@ async def test_table_csv_with_invalid_labels():
)
await ds.invoke_startup()
db = ds.add_memory_database("db_2214")
await db.execute_write_script(
"""
await db.execute_write_script("""
create table t1 (id integer primary key, name text);
insert into t1 (id, name) values (1, 'one');
insert into t1 (id, name) values (2, 'two');
@ -124,8 +115,7 @@ async def test_table_csv_with_invalid_labels():
insert into maintable (id, fk_integer, fk_text) values (1, 1, 'a');
insert into maintable (id, fk_integer, fk_text) values (2, 3, 'b'); -- invalid fk_integer
insert into maintable (id, fk_integer, fk_text) values (3, 2, 'c'); -- invalid fk_text
"""
)
""")
response = await ds.client.get("/db_2214/maintable.csv?_labels=1")
assert response.status_code == 200
assert response.text == (

7
tests/test_html.py
View file

@ -620,14 +620,11 @@ async def test_urlify_custom_queries(ds_client):
response = await ds_client.get(path)
assert response.status_code == 200
soup = Soup(response.content, "html.parser")
assert (
"""<td class="col-user_url">
assert """<td class="col-user_url">
<a href="https://twitter.com/simonw">
https://twitter.com/simonw
</a>
</td>"""
== soup.find("td", {"class": "col-user_url"}).prettify().strip()
)
</td>""" == soup.find("td", {"class": "col-user_url"}).prettify().strip()
@pytest.mark.asyncio

12
tests/test_internals_database.py
View file

@ -747,19 +747,15 @@ async def test_replace_database(tmpdir):
path1 = str(tmpdir / "data1.db")
(tmpdir / "two").mkdir()
path2 = str(tmpdir / "two" / "data1.db")
sqlite3.connect(path1).executescript(
"""
sqlite3.connect(path1).executescript("""
create table t (id integer primary key);
insert into t (id) values (1);
insert into t (id) values (2);
"""
)
sqlite3.connect(path2).executescript(
"""
""")
sqlite3.connect(path2).executescript("""
create table t (id integer primary key);
insert into t (id) values (1);
"""
)
""")
datasette = Datasette([path1])
db = datasette.get_database("data1")
count = (await db.execute("select count(*) from t")).first()[0]

10
tests/test_plugins.py
View file

@ -233,9 +233,7 @@ async def test_hook_render_cell_pks_compound_pk(ds_client):
@pytest.mark.asyncio
async def test_hook_render_cell_pks_rowid_table(ds_client):
"""pks should be ["rowid"] for a table with no explicit primary key"""
response = await ds_client.get(
"/fixtures/no_primary_key?content=RENDER_CELL_DEMO"
)
response = await ds_client.get("/fixtures/no_primary_key?content=RENDER_CELL_DEMO")
soup = Soup(response.text, "html.parser")
td = soup.find("td", {"class": "col-content"})
data = json.loads(td.string)
@ -457,14 +455,12 @@ def view_names_client(tmp_path_factory):
):
(templates / template).write_text("view_name:{{ view_name }}", "utf-8")
(plugins / "extra_vars.py").write_text(
textwrap.dedent(
"""
textwrap.dedent("""
from datasette import hookimpl
@hookimpl
def extra_template_vars(view_name):
return {"view_name": view_name}
"""
),
"""),
"utf-8",
)
db_path = str(tmpdir / "fixtures.db")

14
tests/test_publish_cloudrun.py
View file

@ -231,16 +231,12 @@ def test_publish_cloudrun_plugin_secrets(
with open("test.db", "w") as fp:
fp.write("data")
with open("metadata.yml", "w") as fp:
fp.write(
textwrap.dedent(
"""
fp.write(textwrap.dedent("""
title: Hello from metadata YAML
plugins:
datasette-auth-github:
foo: bar
"""
).strip()
)
""").strip())
result = runner.invoke(
cli.cli,
[
@ -333,8 +329,7 @@ def test_publish_cloudrun_apt_get_install(
.split("\n====================\n")[0]
.strip()
)
expected = textwrap.dedent(
r"""
expected = textwrap.dedent(r"""
FROM python:3.11.0-slim-bullseye
COPY . /app
WORKDIR /app
@ -350,8 +345,7 @@ def test_publish_cloudrun_apt_get_install(
ENV PORT 8001
EXPOSE 8001
CMD datasette serve --host 0.0.0.0 -i test.db --cors --inspect-file inspect-data.json --setting force_https_urls on --port $PORT
"""
).strip()
""").strip()
assert expected == dockerfile

6
tests/test_routes.py
View file

@ -63,12 +63,10 @@ async def ds_with_route():
ds.remove_database("_memory")
db = Database(ds, is_memory=True, memory_name="route-name-db")
ds.add_database(db, name="original-name", route="custom-route-name")
await db.execute_write_script(
"""
await db.execute_write_script("""
create table if not exists t (id integer primary key);
insert or replace into t (id) values (1);
"""
)
""")
return ds

8
tests/test_table_api.py
View file

@ -1243,9 +1243,7 @@ async def test_paginate_using_link_header(ds_client, qs):
reason="generated columns were added in SQLite 3.31.0",
)
def test_generated_columns_are_visible_in_datasette():
with make_app_client(
extra_databases={
"generated.db": """
with make_app_client(extra_databases={"generated.db": """
CREATE TABLE generated_columns (
body TEXT,
id INT GENERATED ALWAYS AS (json_extract(body, '$.number')) STORED,
@ -1253,9 +1251,7 @@ def test_generated_columns_are_visible_in_datasette():
);
INSERT INTO generated_columns (body) VALUES (
'{"number": 1, "string": "This is a string"}'
);"""
}
) as client:
);"""}) as client:
response = client.get("/generated/generated_columns.json?_shape=array")
assert response.json == [
{

22
tests/test_utils.py
View file

@ -201,9 +201,7 @@ def test_detect_fts(open_quote, close_quote):
CREATE VIEW Test_View AS SELECT * FROM Dumb_Table;
CREATE VIRTUAL TABLE {open}Street_Tree_List_fts{close} USING FTS4 ("qAddress", "qCaretaker", "qSpecies", content={open}Street_Tree_List{close});
CREATE VIRTUAL TABLE r USING rtree(a, b, c);
""".format(
open=open_quote, close=close_quote
)
""".format(open=open_quote, close=close_quote)
conn = utils.sqlite3.connect(":memory:")
conn.executescript(sql)
assert None is utils.detect_fts(conn, "Dumb_Table")
@ -220,9 +218,7 @@ def test_detect_fts_different_table_names(table):
"qSpecies" TEXT
);
CREATE VIRTUAL TABLE [{table}_fts] USING FTS4 ("qSpecies", content="{table}");
""".format(
table=table
)
""".format(table=table)
conn = utils.sqlite3.connect(":memory:")
conn.executescript(sql)
assert "{table}_fts".format(table=table) == utils.detect_fts(conn, table)
@ -347,27 +343,21 @@ def test_compound_keys_after_sql():
((a > :p0)
or
(a = :p0 and b > :p1))
""".strip() == utils.compound_keys_after_sql(
["a", "b"]
)
""".strip() == utils.compound_keys_after_sql(["a", "b"])
assert """
((a > :p0)
or
(a = :p0 and b > :p1)
or
(a = :p0 and b = :p1 and c > :p2))
""".strip() == utils.compound_keys_after_sql(
["a", "b", "c"]
)
""".strip() == utils.compound_keys_after_sql(["a", "b", "c"])
def test_table_columns():
conn = sqlite3.connect(":memory:")
conn.executescript(
"""
conn.executescript("""
create table places (id integer primary key, name text, bob integer)
"""
)
""")
assert ["id", "name", "bob"] == utils.table_columns(conn, "places")

6
tests/test_utils_permissions.py
View file

@ -497,16 +497,14 @@ async def test_actor_actor_id_action_parameters_available(db):
def plugin_using_all_parameters() -> Callable[[str], PermissionSQL]:
def provider(action: str) -> PermissionSQL:
return PermissionSQL(
"""
return PermissionSQL("""
SELECT NULL AS parent, NULL AS child, 1 AS allow,
'Actor ID: ' || COALESCE(:actor_id, 'null') ||
', Actor JSON: ' || COALESCE(:actor, 'null') ||
', Action: ' || :action AS reason
WHERE :actor_id = 'test_user' AND :action = 'view-table'
AND json_extract(:actor, '$.role') = 'admin'
"""
)
""")
return provider

4
tests/test_write_wrapper.py
View file

@ -471,7 +471,9 @@ async def test_write_wrapper_set_authorizer(datasette, actor, table, should_deny
),
request=request,
)
result = await db.execute(f"select value from {table} order by rowid desc limit 1")
result = await db.execute(
f"select value from {table} order by rowid desc limit 1"
)
assert result.rows[0][0] == "test"
finally:
pm.unregister(name="test_set_authorizer")