diff --git a/datasette/app.py b/datasette/app.py
index 96683895..56b89789 100644
--- a/datasette/app.py
+++ b/datasette/app.py
@@ -1029,8 +1029,8 @@ class Datasette:
)
@staticmethod
- def _query_row_to_dict(row):
- return stored_queries.query_row_to_dict(row)
+ def _query_row_to_stored_query(row) -> stored_queries.StoredQuery | None:
+ return stored_queries.query_row_to_stored_query(row)
@staticmethod
def _query_options_json(options):
@@ -1038,28 +1038,28 @@ class Datasette:
async def add_query(
self,
- database,
- name,
- sql,
+ database: str,
+ name: str,
+ sql: str,
*,
- title=None,
- description=None,
- description_html=None,
- hide_sql=False,
- fragment=None,
- parameters=None,
- is_write=False,
- is_private=False,
- is_trusted=False,
- source="plugin",
- owner_id=None,
- on_success_message=None,
- on_success_message_sql=None,
- on_success_redirect=None,
- on_error_message=None,
- on_error_redirect=None,
- replace=True,
- ):
+ title: str | None = None,
+ description: str | None = None,
+ description_html: str | None = None,
+ hide_sql: bool = False,
+ fragment: str | None = None,
+ parameters: Iterable[str] | None = None,
+ is_write: bool = False,
+ is_private: bool = False,
+ is_trusted: bool = False,
+ source: str = "plugin",
+ owner_id: str | None = None,
+ on_success_message: str | None = None,
+ on_success_message_sql: str | None = None,
+ on_success_redirect: str | None = None,
+ on_error_message: str | None = None,
+ on_error_redirect: str | None = None,
+ replace: bool = True,
+ ) -> None:
return await stored_queries.add_query(
self,
database,
@@ -1086,8 +1086,8 @@ class Datasette:
async def update_query(
self,
- database,
- name,
+ database: str,
+ name: str,
*,
sql=stored_queries.UNCHANGED,
title=stored_queries.UNCHANGED,
@@ -1106,7 +1106,7 @@ class Datasette:
on_success_redirect=stored_queries.UNCHANGED,
on_error_message=stored_queries.UNCHANGED,
on_error_redirect=stored_queries.UNCHANGED,
- ):
+ ) -> None:
return await stored_queries.update_query(
self,
database,
@@ -1130,24 +1130,28 @@ class Datasette:
on_error_redirect=on_error_redirect,
)
- async def remove_query(self, database, name, source=None):
+ async def remove_query(
+ self, database: str, name: str, source: str | None = None
+ ) -> None:
return await stored_queries.remove_query(self, database, name, source=source)
- async def get_query(self, database, name):
+ async def get_query(
+ self, database: str, name: str
+ ) -> stored_queries.StoredQuery | None:
return await stored_queries.get_query(self, database, name)
async def count_queries(
self,
- database=None,
+ database: str | None = None,
*,
- actor=None,
- q=None,
- is_write=None,
- is_private=None,
- is_trusted=None,
- source=None,
- owner_id=None,
- ):
+ actor: dict[str, Any] | None = None,
+ q: str | None = None,
+ is_write: bool | None = None,
+ is_private: bool | None = None,
+ is_trusted: bool | None = None,
+ source: str | None = None,
+ owner_id: str | None = None,
+ ) -> int:
return await stored_queries.count_queries(
self,
database,
@@ -1162,19 +1166,19 @@ class Datasette:
async def list_queries(
self,
- database=None,
+ database: str | None = None,
*,
- actor=None,
- limit=50,
- cursor=None,
- q=None,
- is_write=None,
- is_private=None,
- is_trusted=None,
- source=None,
- owner_id=None,
- include_private=False,
- ):
+ actor: dict[str, Any] | None = None,
+ limit: int = 50,
+ cursor: str | None = None,
+ q: str | None = None,
+ is_write: bool | None = None,
+ is_private: bool | None = None,
+ is_trusted: bool | None = None,
+ source: str | None = None,
+ owner_id: str | None = None,
+ include_private: bool = False,
+ ) -> stored_queries.StoredQueryPage:
return await stored_queries.list_queries(
self,
database,
diff --git a/datasette/stored_queries.py b/datasette/stored_queries.py
index a28b71bf..bcfdfdb4 100644
--- a/datasette/stored_queries.py
+++ b/datasette/stored_queries.py
@@ -1,6 +1,8 @@
from __future__ import annotations
+from dataclasses import dataclass
import json
+from typing import Any, Iterable
from .resources import TableResource
from .utils import named_parameters, sqlite3, tilde_encode, urlsafe_components
@@ -19,7 +21,76 @@ QUERY_OPTION_FIELDS = (
)
-async def save_queries_from_config(datasette):
+@dataclass
+class StoredQuery:
+ database: str
+ name: str
+ sql: str
+ title: str | None
+ description: str | None
+ description_html: str | None
+ hide_sql: bool
+ fragment: str | None
+ parameters: list[str]
+ is_write: bool
+ is_private: bool
+ is_trusted: bool
+ source: str
+ owner_id: str | None
+ on_success_message: str | None
+ on_success_message_sql: str | None
+ on_success_redirect: str | None
+ on_error_message: str | None
+ on_error_redirect: str | None
+ private: bool | None = None
+
+
+@dataclass
+class StoredQueryPage:
+ queries: list[StoredQuery]
+ next: str | None
+ has_more: bool
+ limit: int
+
+
+def stored_query_to_dict(query: StoredQuery) -> dict[str, Any]:
+ data = {
+ "database": query.database,
+ "name": query.name,
+ "sql": query.sql,
+ "title": query.title,
+ "description": query.description,
+ "description_html": query.description_html,
+ "hide_sql": query.hide_sql,
+ "fragment": query.fragment,
+ "params": list(query.parameters),
+ "parameters": list(query.parameters),
+ "is_write": query.is_write,
+ "is_private": query.is_private,
+ "is_trusted": query.is_trusted,
+ "source": query.source,
+ "owner_id": query.owner_id,
+ "on_success_message": query.on_success_message,
+ "on_success_message_sql": query.on_success_message_sql,
+ "on_success_redirect": query.on_success_redirect,
+ "on_error_message": query.on_error_message,
+ "on_error_redirect": query.on_error_redirect,
+ }
+ if query.private is not None:
+ data["private"] = query.private
+ return data
+
+
+def stored_query_page_to_dict(page: StoredQueryPage) -> dict[str, Any]:
+ return {
+ "queries": [stored_query_to_dict(query) for query in page.queries],
+ "next": page.next,
+ "has_more": page.has_more,
+ "limit": page.limit,
+ }
+
+
+async def save_queries_from_config(datasette: Any) -> None:
# Apply configured query entries from datasette.yaml to the internal table.
await datasette.get_internal_database().execute_write(
"DELETE FROM queries WHERE source = 'config'"
@@ -50,36 +121,38 @@ async def save_queries_from_config(datasette):
)
-def query_row_to_dict(row):
+def query_row_to_stored_query(
+ row: Any, private: bool | None = None
+) -> StoredQuery | None:
if row is None:
return None
parameters = json.loads(row["parameters"] or "[]")
options = json.loads(row["options"] or "{}")
- return {
- "database": row["database_name"],
- "name": row["name"],
- "sql": row["sql"],
- "title": row["title"],
- "description": row["description"],
- "description_html": row["description_html"],
- "hide_sql": bool(options.get("hide_sql")),
- "fragment": options.get("fragment"),
- "params": parameters,
- "parameters": parameters,
- "is_write": bool(row["is_write"]),
- "is_private": bool(row["is_private"]),
- "is_trusted": bool(row["is_trusted"]),
- "source": row["source"],
- "owner_id": row["owner_id"],
- "on_success_message": options.get("on_success_message"),
- "on_success_message_sql": options.get("on_success_message_sql"),
- "on_success_redirect": options.get("on_success_redirect"),
- "on_error_message": options.get("on_error_message"),
- "on_error_redirect": options.get("on_error_redirect"),
- }
+ return StoredQuery(
+ database=row["database_name"],
+ name=row["name"],
+ sql=row["sql"],
+ title=row["title"],
+ description=row["description"],
+ description_html=row["description_html"],
+ hide_sql=bool(options.get("hide_sql")),
+ fragment=options.get("fragment"),
+ parameters=parameters,
+ is_write=bool(row["is_write"]),
+ is_private=bool(row["is_private"]),
+ is_trusted=bool(row["is_trusted"]),
+ source=row["source"],
+ owner_id=row["owner_id"],
+ on_success_message=options.get("on_success_message"),
+ on_success_message_sql=options.get("on_success_message_sql"),
+ on_success_redirect=options.get("on_success_redirect"),
+ on_error_message=options.get("on_error_message"),
+ on_error_redirect=options.get("on_error_redirect"),
+ private=private,
+ )
-def query_options_json(options):
+def query_options_json(options: dict[str, Any]) -> str:
options_dict = {}
for field in QUERY_OPTION_FIELDS:
value = options.get(field)
@@ -92,29 +165,29 @@ def query_options_json(options):
async def add_query(
- datasette,
- database,
- name,
- sql,
+ datasette: Any,
+ database: str,
+ name: str,
+ sql: str,
*,
- title=None,
- description=None,
- description_html=None,
- hide_sql=False,
- fragment=None,
- parameters=None,
- is_write=False,
- is_private=False,
- is_trusted=False,
- source="plugin",
- owner_id=None,
- on_success_message=None,
- on_success_message_sql=None,
- on_success_redirect=None,
- on_error_message=None,
- on_error_redirect=None,
- replace=True,
-):
+ title: str | None = None,
+ description: str | None = None,
+ description_html: str | None = None,
+ hide_sql: bool = False,
+ fragment: str | None = None,
+ parameters: Iterable[str] | None = None,
+ is_write: bool = False,
+ is_private: bool = False,
+ is_trusted: bool = False,
+ source: str = "plugin",
+ owner_id: str | None = None,
+ on_success_message: str | None = None,
+ on_success_message_sql: str | None = None,
+ on_success_redirect: str | None = None,
+ on_error_message: str | None = None,
+ on_error_redirect: str | None = None,
+ replace: bool = True,
+) -> None:
parameters_json = json.dumps(list(parameters or []))
options_json = query_options_json(
{
@@ -170,9 +243,9 @@ async def add_query(
async def update_query(
- datasette,
- database,
- name,
+ datasette: Any,
+ database: str,
+ name: str,
*,
sql=UNCHANGED,
title=UNCHANGED,
@@ -191,7 +264,7 @@ async def update_query(
on_success_redirect=UNCHANGED,
on_error_message=UNCHANGED,
on_error_redirect=UNCHANGED,
-):
+) -> None:
fields = {
"sql": sql,
"title": title,
@@ -263,7 +336,9 @@ async def update_query(
)
-async def remove_query(datasette, database, name, source=None):
+async def remove_query(
+ datasette: Any, database: str, name: str, source: str | None = None
+) -> None:
sql = "DELETE FROM queries WHERE database_name = ? AND name = ?"
params = [database, name]
if source is not None:
@@ -272,7 +347,7 @@ async def remove_query(datasette, database, name, source=None):
await datasette.get_internal_database().execute_write(sql, params)
-async def get_query(datasette, database, name):
+async def get_query(datasette: Any, database: str, name: str) -> StoredQuery | None:
rows = await datasette.get_internal_database().execute(
"""
SELECT * FROM queries
@@ -280,21 +355,21 @@ async def get_query(datasette, database, name):
""",
[database, name],
)
- return query_row_to_dict(rows.first())
+ return query_row_to_stored_query(rows.first())
async def count_queries(
- datasette,
- database=None,
+ datasette: Any,
+ database: str | None = None,
*,
- actor=None,
- q=None,
- is_write=None,
- is_private=None,
- is_trusted=None,
- source=None,
- owner_id=None,
-):
+ actor: dict[str, Any] | None = None,
+ q: str | None = None,
+ is_write: bool | None = None,
+ is_private: bool | None = None,
+ is_trusted: bool | None = None,
+ source: str | None = None,
+ owner_id: str | None = None,
+) -> int:
allowed_sql, allowed_params = await datasette.allowed_resources_sql(
action="view-query",
actor=actor,
@@ -354,20 +429,20 @@ async def count_queries(
async def list_queries(
- datasette,
- database=None,
+ datasette: Any,
+ database: str | None = None,
*,
- actor=None,
- limit=50,
- cursor=None,
- q=None,
- is_write=None,
- is_private=None,
- is_trusted=None,
- source=None,
- owner_id=None,
- include_private=False,
-):
+ actor: dict[str, Any] | None = None,
+ limit: int = 50,
+ cursor: str | None = None,
+ q: str | None = None,
+ is_write: bool | None = None,
+ is_private: bool | None = None,
+ is_trusted: bool | None = None,
+ source: str | None = None,
+ owner_id: str | None = None,
+ include_private: bool = False,
+) -> StoredQueryPage:
limit = min(max(1, int(limit)), 1000)
allowed_sql, allowed_params = await datasette.allowed_resources_sql(
action="view-query",
@@ -480,9 +555,10 @@ async def list_queries(
queries = []
for row in rows:
- query = query_row_to_dict(row)
- if include_private:
- query["private"] = bool(row["private"])
+ query = query_row_to_stored_query(
+ row, private=bool(row["private"]) if include_private else None
+ )
+ assert query is not None
queries.append(query)
next_token = None
@@ -499,17 +575,23 @@ async def list_queries(
tilde_encode(last_row["sort_key"]),
tilde_encode(last_row["name"]),
)
- return {
- "queries": queries,
- "next": next_token,
- "has_more": has_more,
- "limit": limit,
- }
+ return StoredQueryPage(
+ queries=queries,
+ next=next_token,
+ has_more=has_more,
+ limit=limit,
+ )
async def ensure_query_write_permissions(
- datasette, database, sql, *, actor=None, params=None, analysis=None
-):
+ datasette: Any,
+ database: str,
+ sql: str,
+ *,
+ actor: dict[str, Any] | None = None,
+ params: dict[str, Any] | None = None,
+ analysis: Any = None,
+) -> Any:
write_actions = {
"insert": "insert-row",
"update": "update-row",
diff --git a/datasette/views/database.py b/datasette/views/database.py
index 98ca989c..b558b002 100644
--- a/datasette/views/database.py
+++ b/datasette/views/database.py
@@ -13,6 +13,7 @@ import textwrap
from datasette.events import AlterTableEvent, CreateTableEvent, InsertRowsEvent
from datasette.database import QueryInterrupted
from datasette.resources import DatabaseResource, QueryResource
+from datasette.stored_queries import stored_query_to_dict
from datasette.utils import (
add_cors_headers,
await_me_maybe,
@@ -99,8 +100,8 @@ class DatabaseView(View):
limit=5,
include_private=True,
)
- stored_queries = queries_page["queries"]
- queries_more = queries_page["has_more"]
+ stored_queries = queries_page.queries
+ queries_more = queries_page.has_more
queries_count = (
await datasette.count_queries(database, actor=request.actor)
if queries_more
@@ -136,7 +137,7 @@ class DatabaseView(View):
"tables": tables,
"hidden_count": len([t for t in tables if t["hidden"]]),
"views": sql_views,
- "queries": stored_queries,
+ "queries": [stored_query_to_dict(query) for query in stored_queries],
"queries_more": queries_more,
"queries_count": queries_count,
"allow_execute_sql": allow_execute_sql,
@@ -447,7 +448,7 @@ class QueryView(View):
if not await datasette.allowed(
action="view-query",
- resource=QueryResource(database=db.name, query=stored_query["name"]),
+ resource=QueryResource(database=db.name, query=stored_query.name),
actor=request.actor,
):
raise Forbidden("You do not have permission to view this query")
@@ -480,20 +481,18 @@ class QueryView(View):
or request.args.get("_json")
or params.get("_json")
)
- params_for_query = MagicParameters(
- stored_query["sql"], params, request, datasette
- )
+ params_for_query = MagicParameters(stored_query.sql, params, request, datasette)
await params_for_query.execute_params()
ok = None
redirect_url = None
try:
cursor = await db.execute_write(
- stored_query["sql"], params_for_query, request=request
+ stored_query.sql, params_for_query, request=request
)
# success message can come from on_success_message or on_success_message_sql
message = None
message_type = datasette.INFO
- on_success_message_sql = stored_query.get("on_success_message_sql")
+ on_success_message_sql = stored_query.on_success_message_sql
if on_success_message_sql:
try:
message_result = (
@@ -505,18 +504,19 @@ class QueryView(View):
message = "Error running on_success_message_sql: {}".format(ex)
message_type = datasette.ERROR
if not message:
- message = stored_query.get(
- "on_success_message"
- ) or "Query executed, {} row{} affected".format(
- cursor.rowcount, "" if cursor.rowcount == 1 else "s"
+ message = (
+ stored_query.on_success_message
+ or "Query executed, {} row{} affected".format(
+ cursor.rowcount, "" if cursor.rowcount == 1 else "s"
+ )
)
- redirect_url = stored_query.get("on_success_redirect")
+ redirect_url = stored_query.on_success_redirect
ok = True
except Exception as ex:
- message = stored_query.get("on_error_message") or str(ex)
+ message = stored_query.on_error_message or str(ex)
message_type = datasette.ERROR
- redirect_url = stored_query.get("on_error_redirect")
+ redirect_url = stored_query.on_error_redirect
ok = False
if should_return_json:
return Response.json(
@@ -562,7 +562,7 @@ class QueryView(View):
)
if stored_query is None:
raise
- stored_query_write = bool(stored_query.get("is_write"))
+ stored_query_write = stored_query.is_write
private = False
if stored_query:
@@ -570,7 +570,7 @@ class QueryView(View):
visible, private = await datasette.check_visibility(
request.actor,
action="view-query",
- resource=QueryResource(database=database, query=stored_query["name"]),
+ resource=QueryResource(database=database, query=stored_query.name),
)
if not visible:
raise Forbidden("You do not have permission to view this query")
@@ -591,14 +591,14 @@ class QueryView(View):
sql = None
if stored_query:
- sql = stored_query["sql"]
+ sql = stored_query.sql
elif "sql" in params:
sql = params.pop("sql")
# Extract any :named parameters
named_parameters = []
- if stored_query and stored_query.get("params"):
- named_parameters = stored_query["params"]
+ if stored_query and stored_query.parameters:
+ named_parameters = stored_query.parameters
if not named_parameters and sql:
named_parameters = derive_named_parameters(sql)
named_parameter_values = {
@@ -686,7 +686,7 @@ class QueryView(View):
columns=columns,
rows=rows,
sql=sql,
- query_name=stored_query["name"] if stored_query else None,
+ query_name=stored_query.name if stored_query else None,
database=database,
table=None,
request=request,
@@ -721,7 +721,7 @@ class QueryView(View):
if stored_query:
templates.insert(
0,
- f"query-{to_css_class(database)}-{to_css_class(stored_query['name'])}.html",
+ f"query-{to_css_class(database)}-{to_css_class(stored_query.name)}.html",
)
environment = datasette.get_jinja_environment(request)
@@ -740,7 +740,7 @@ class QueryView(View):
)
metadata = await datasette.get_database_metadata(database)
if stored_query:
- metadata = dict(stored_query)
+ metadata = stored_query_to_dict(stored_query)
metadata.pop("source", None)
renderers = {}
@@ -775,7 +775,7 @@ class QueryView(View):
)
show_hide_hidden = ""
- if stored_query and stored_query.get("hide_sql"):
+ if stored_query and stored_query.hide_sql:
if bool(params.get("_show_sql")):
show_hide_link = path_with_removed_args(request, {"_show_sql"})
show_hide_text = "hide"
@@ -843,7 +843,7 @@ class QueryView(View):
datasette=datasette,
actor=request.actor,
database=database,
- query_name=stored_query["name"] if stored_query else None,
+ query_name=stored_query.name if stored_query else None,
request=request,
sql=sql,
params=params,
@@ -863,7 +863,7 @@ class QueryView(View):
"sql": sql,
"params": params,
},
- stored_query=stored_query["name"] if stored_query else None,
+ stored_query=stored_query.name if stored_query else None,
private=private,
stored_query_write=stored_query_write,
db_is_immutable=not db.is_mutable,
@@ -907,7 +907,7 @@ class QueryView(View):
datasette,
request,
database=database,
- query_name=stored_query["name"] if stored_query else None,
+ query_name=stored_query.name if stored_query else None,
),
query_actions=query_actions,
),
diff --git a/datasette/views/query_helpers.py b/datasette/views/query_helpers.py
index de732431..46d71b8e 100644
--- a/datasette/views/query_helpers.py
+++ b/datasette/views/query_helpers.py
@@ -2,6 +2,7 @@ import json
import re
from datasette.resources import DatabaseResource, TableResource
+from datasette.stored_queries import StoredQuery
from datasette.utils import (
named_parameters as derive_named_parameters,
escape_sqlite,
@@ -281,18 +282,18 @@ async def _prepare_execute_write(datasette, db, sql, params, actor):
return parameter_names, params, analysis
-async def _ensure_stored_query_execution_permissions(datasette, db, query, actor):
- if query.get("is_trusted"):
+async def _ensure_stored_query_execution_permissions(
+ datasette, db, query: StoredQuery, actor
+):
+ if query.is_trusted:
return
- if query.get("is_write"):
+ if query.is_write:
await datasette.ensure_permission(
action="execute-write-sql",
resource=DatabaseResource(db.name),
actor=actor,
)
- await datasette.ensure_query_write_permissions(
- db.name, query["sql"], actor=actor
- )
+ await datasette.ensure_query_write_permissions(db.name, query.sql, actor=actor)
else:
await datasette.ensure_permission(
action="execute-sql",
@@ -482,7 +483,7 @@ async def _prepare_query_create(datasette, request, db, data):
}
-async def _prepare_query_update(datasette, request, db, existing, update):
+async def _prepare_query_update(datasette, request, db, existing: StoredQuery, update):
invalid_keys = set(update) - _query_update_fields
if invalid_keys:
raise QueryValidationError(
@@ -490,8 +491,8 @@ async def _prepare_query_update(datasette, request, db, existing, update):
)
update = _apply_query_data_types(update)
- sql = update.get("sql", existing["sql"])
- query_is_write = existing["is_write"]
+ sql = update.get("sql", existing.sql)
+ query_is_write = existing.is_write
derived = _derived_query_parameters(sql)
parameters = None
diff --git a/datasette/views/stored_queries.py b/datasette/views/stored_queries.py
index 1a2c5d00..8c4e849e 100644
--- a/datasette/views/stored_queries.py
+++ b/datasette/views/stored_queries.py
@@ -1,6 +1,7 @@
from urllib.parse import parse_qsl, urlencode
from datasette.resources import DatabaseResource, QueryResource
+from datasette.stored_queries import stored_query_to_dict
from datasette.utils import sqlite3, tilde_decode
from datasette.utils.asgi import Response
@@ -100,7 +101,7 @@ class QueryListView(BaseView):
)
query_list_path = self.query_list_path(database)
next_url = None
- if page["next"]:
+ if page.next:
pairs = [
(key, value)
for key, value in parse_qsl(
@@ -108,7 +109,7 @@ class QueryListView(BaseView):
)
if key != "_next"
]
- pairs.append(("_next", page["next"]))
+ pairs.append(("_next", page.next))
next_url = "{}?{}".format(
query_list_path,
urlencode(pairs),
@@ -194,13 +195,13 @@ class QueryListView(BaseView):
"database_color": (
self.ds.get_database(database).color if database is not None else None
),
- "queries": page["queries"],
- "next": page["next"],
+ "queries": page.queries,
+ "next": page.next,
"next_url": next_url,
- "has_more": page["has_more"],
- "limit": page["limit"],
- "show_private_note": any(query["is_private"] for query in page["queries"]),
- "show_trusted_note": any(query["is_trusted"] for query in page["queries"]),
+ "has_more": page.has_more,
+ "limit": page.limit,
+ "show_private_note": any(query.is_private for query in page.queries),
+ "show_trusted_note": any(query.is_trusted for query in page.queries),
"query_list_path": query_list_path,
"show_database": database is None,
"facets": facets,
@@ -213,7 +214,12 @@ class QueryListView(BaseView):
},
}
if format_ == "json":
- return Response.json(data)
+ return Response.json(
+ {
+ **data,
+ "queries": [stored_query_to_dict(query) for query in page.queries],
+ }
+ )
return await self.render(
["query_list.html"],
request,
@@ -374,8 +380,11 @@ class QueryStoreView(QueryCreateView):
return _error([str(ex)], 400)
query = await self.ds.get_query(db.name, name)
+ assert query is not None
if is_json:
- return Response.json({"ok": True, "query": query}, status=201)
+ return Response.json(
+ {"ok": True, "query": stored_query_to_dict(query)}, status=201
+ )
self.ds.add_message(request, "Query saved", self.ds.INFO)
return Response.redirect(self.ds.urls.path(self.ds.urls.table(db.name, name)))
@@ -395,7 +404,7 @@ class QueryDefinitionView(BaseView):
actor=request.actor,
):
return _error(["Permission denied"], 403)
- return Response.json({"ok": True, "query": query})
+ return Response.json({"ok": True, "query": stored_query_to_dict(query)})
class QueryUpdateView(BaseView):
@@ -413,7 +422,7 @@ class QueryUpdateView(BaseView):
actor=request.actor,
):
return _error(["Permission denied: need update-query"], 403)
- if existing.get("is_trusted"):
+ if existing.is_trusted:
return _error(["Trusted queries cannot be updated using the API"], 403)
try:
@@ -444,10 +453,12 @@ class QueryUpdateView(BaseView):
await self.ds.update_query(db.name, query_name, **update_kwargs)
if data.get("return"):
+ query = await self.ds.get_query(db.name, query_name)
+ assert query is not None
return Response.json(
{
"ok": True,
- "query": await self.ds.get_query(db.name, query_name),
+ "query": stored_query_to_dict(query),
}
)
return Response.json({"ok": True})
diff --git a/docs/internals.rst b/docs/internals.rst
index 66724aa9..4980ee8b 100644
--- a/docs/internals.rst
+++ b/docs/internals.rst
@@ -1039,11 +1039,11 @@ Example:
await .get_query(database, name)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-Returns a stored query dictionary, or ``None`` if the query does not exist.
+Returns a ``StoredQuery`` dataclass instance, or ``None`` if the query does not exist.
-The dictionary contains ``database``, ``name``, ``sql``, ``title``, ``description``, ``description_html``, ``hide_sql``, ``fragment``, ``parameters``, ``params``, ``is_write``, ``is_private``, ``is_trusted``, ``source``, ``owner_id``, ``on_success_message``, ``on_success_message_sql``, ``on_success_redirect``, ``on_error_message`` and ``on_error_redirect``.
+``StoredQuery`` has the following attributes: ``database``, ``name``, ``sql``, ``title``, ``description``, ``description_html``, ``hide_sql``, ``fragment``, ``parameters``, ``is_write``, ``is_private``, ``is_trusted``, ``source``, ``owner_id``, ``on_success_message``, ``on_success_message_sql``, ``on_success_redirect``, ``on_error_message`` and ``on_error_redirect``.
-``parameters`` and ``params`` contain the same list of explicit parameter names.
+``parameters`` is a list of explicit parameter names.
.. _datasette_list_queries:
@@ -1087,12 +1087,12 @@ Lists stored queries visible to the specified actor.
``owner_id`` - string, optional
Filter by owner actor ID.
``include_private`` - boolean, optional
- Set to ``True`` to include a ``private`` boolean in each returned query dictionary indicating if anonymous users would be unable to view that query.
+ Set to ``True`` to populate a ``private`` boolean on each returned ``StoredQuery`` indicating if anonymous users would be unable to view that query.
-The return value is a dictionary with these keys:
+The return value is a ``StoredQueryPage`` dataclass instance with these attributes:
-``queries`` - list of dictionaries
- Stored query dictionaries, in the same format returned by :ref:`datasette_get_query`.
+``queries`` - list of StoredQuery instances
+ Stored queries in the same format returned by :ref:`datasette_get_query`.
``next`` - string or None
Pagination cursor for the next page, if one exists.
``has_more`` - boolean
diff --git a/tests/test_queries.py b/tests/test_queries.py
index 70fb7a03..59fab8c0 100644
--- a/tests/test_queries.py
+++ b/tests/test_queries.py
@@ -4,6 +4,7 @@ import pytest
from datasette.app import Datasette
from datasette.resources import DatabaseResource, QueryResource
+from datasette.stored_queries import StoredQuery, StoredQueryPage
from datasette.utils.asgi import Forbidden
@@ -87,38 +88,41 @@ async def test_add_get_and_remove_query():
}
query = await ds.get_query("data", "top_customers")
- assert query == {
- "database": "data",
- "name": "top_customers",
- "sql": "select * from customers where region = :region",
- "title": "Top customers",
- "description": "Customers by region",
- "description_html": None,
- "hide_sql": True,
- "fragment": "chart",
- "params": ["region"],
- "parameters": ["region"],
- "is_write": False,
- "is_private": False,
- "is_trusted": True,
- "source": "user",
- "owner_id": "alice",
- "on_success_message": None,
- "on_success_message_sql": None,
- "on_success_redirect": None,
- "on_error_message": None,
- "on_error_redirect": None,
- }
+ assert query == StoredQuery(
+ database="data",
+ name="top_customers",
+ sql="select * from customers where region = :region",
+ title="Top customers",
+ description="Customers by region",
+ description_html=None,
+ hide_sql=True,
+ fragment="chart",
+ parameters=["region"],
+ is_write=False,
+ is_private=False,
+ is_trusted=True,
+ source="user",
+ owner_id="alice",
+ on_success_message=None,
+ on_success_message_sql=None,
+ on_success_redirect=None,
+ on_error_message=None,
+ on_error_redirect=None,
+ )
queries_page = await ds.list_queries("data", actor=None)
- assert queries_page["queries"] == [query]
- assert queries_page["next"] is None
+ assert queries_page == StoredQueryPage(
+ queries=[query],
+ next=None,
+ has_more=False,
+ limit=50,
+ )
await ds.remove_query("data", "top_customers")
assert await ds.get_query("data", "top_customers") is None
queries_page = await ds.list_queries("data", actor=None)
- assert queries_page["queries"] == []
- assert queries_page["next"] is None
+ assert queries_page.queries == []
+ assert queries_page.next is None
@pytest.mark.asyncio
@@ -156,13 +160,12 @@ async def test_update_query_only_updates_provided_fields():
)
query = await ds.get_query("data", "redirect")
- assert query["title"] == "Updated"
- assert query["parameters"] == []
- assert query["params"] == []
- assert query["on_success_redirect"] is None
- assert query["sql"] == "select 1"
- assert query["is_private"] is False
- assert query["is_trusted"] is False
+ assert query.title == "Updated"
+ assert query.parameters == []
+ assert query.on_success_redirect is None
+ assert query.sql == "select 1"
+ assert query.is_private is False
+ assert query.is_trusted is False
options_row = (
await ds.get_internal_database().execute(
"""
@@ -198,28 +201,27 @@ async def test_config_queries_imported_to_internal_table():
ds.add_memory_database("query_config", name="data")
await ds.invoke_startup()
- assert await ds.get_query("data", "configured") == {
- "database": "data",
- "name": "configured",
- "sql": "select :name as name",
- "title": "Configured query",
- "description": None,
- "description_html": "Configured HTML
",
- "hide_sql": False,
- "fragment": None,
- "params": ["name"],
- "parameters": ["name"],
- "is_write": False,
- "is_private": False,
- "is_trusted": True,
- "source": "config",
- "owner_id": None,
- "on_success_message": None,
- "on_success_message_sql": "select 'Hello ' || :name",
- "on_success_redirect": None,
- "on_error_message": None,
- "on_error_redirect": None,
- }
+ assert await ds.get_query("data", "configured") == StoredQuery(
+ database="data",
+ name="configured",
+ sql="select :name as name",
+ title="Configured query",
+ description=None,
+ description_html="Configured HTML
",
+ hide_sql=False,
+ fragment=None,
+ parameters=["name"],
+ is_write=False,
+ is_private=False,
+ is_trusted=True,
+ source="config",
+ owner_id=None,
+ on_success_message=None,
+ on_success_message_sql="select 'Hello ' || :name",
+ on_success_redirect=None,
+ on_error_message=None,
+ on_error_redirect=None,
+ )
@pytest.mark.asyncio
@@ -1032,8 +1034,8 @@ async def test_query_update_api_rejects_config_only_fields():
"Invalid keys: description_html, on_success_message_sql"
]
query = await ds.get_query("data", "editable")
- assert query["description_html"] is None
- assert query["on_success_message_sql"] is None
+ assert query.description_html is None
+ assert query.on_success_message_sql is None
@pytest.mark.asyncio
@@ -1072,9 +1074,9 @@ async def test_query_update_api_rejects_trusted_queries_but_internal_update_allo
"Trusted queries cannot be updated using the API"
]
query = await ds.get_query("data", "trusted_report")
- assert query["is_trusted"] is True
- assert query["sql"] == "select 1 as one"
- assert query["title"] == "Original"
+ assert query.is_trusted is True
+ assert query.sql == "select 1 as one"
+ assert query.title == "Original"
await ds.update_query(
"data",
@@ -1083,9 +1085,9 @@ async def test_query_update_api_rejects_trusted_queries_but_internal_update_allo
title="Internal",
)
query = await ds.get_query("data", "trusted_report")
- assert query["is_trusted"] is True
- assert query["sql"] == "select 3 as three"
- assert query["title"] == "Internal"
+ assert query.is_trusted is True
+ assert query.sql == "select 3 as three"
+ assert query.title == "Internal"
@pytest.mark.asyncio