diff --git a/datasette/app.py b/datasette/app.py
index c684eabc..e131ba46 100644
--- a/datasette/app.py
+++ b/datasette/app.py
@@ -387,18 +387,28 @@ class Datasette:
             ).hexdigest()[:6]
         return self._app_css_hash
 
-    def get_canned_queries(self, database_name):
+    async def get_canned_queries(self, database_name, actor):
         queries = self.metadata("queries", database=database_name, fallback=False) or {}
-        names = queries.keys()
-        return [self.get_canned_query(database_name, name) for name in names]
+        for more_queries in pm.hook.canned_queries(
+            datasette=self, database=database_name, actor=actor,
+        ):
+            if callable(more_queries):
+                more_queries = more_queries()
+            if asyncio.iscoroutine(more_queries):
+                more_queries = await more_queries
+            queries.update(more_queries or {})
+        # Fix any {"name": "select ..."} queries to be {"name": {"sql": "select ..."}}
+        for key in queries:
+            if not isinstance(queries[key], dict):
+                queries[key] = {"sql": queries[key]}
+            # Also make sure "name" is available:
+            queries[key]["name"] = key
+        return queries
 
-    def get_canned_query(self, database_name, query_name):
-        queries = self.metadata("queries", database=database_name, fallback=False) or {}
+    async def get_canned_query(self, database_name, query_name, actor):
+        queries = await self.get_canned_queries(database_name, actor)
         query = queries.get(query_name)
         if query:
-            if not isinstance(query, dict):
-                query = {"sql": query}
-            query["name"] = query_name
             return query
 
     def update_with_inherited_metadata(self, metadata):
diff --git a/datasette/default_permissions.py b/datasette/default_permissions.py
index e750acbf..0929a17a 100644
--- a/datasette/default_permissions.py
+++ b/datasette/default_permissions.py
@@ -4,41 +4,42 @@ from datasette.utils import actor_matches_allow
 
 @hookimpl(tryfirst=True)
 def permission_allowed(datasette, actor, action, resource):
-    if action == "permissions-debug":
-        if actor and actor.get("id") == "root":
-            return True
-    elif action == "view-instance":
-        allow = datasette.metadata("allow")
-        if allow is not None:
+    async def inner():
+        if action == "permissions-debug":
+            if actor and actor.get("id") == "root":
+                return True
+        elif action == "view-instance":
+            allow = datasette.metadata("allow")
+            if allow is not None:
+                return actor_matches_allow(actor, allow)
+        elif action == "view-database":
+            database_allow = datasette.metadata("allow", database=resource)
+            if database_allow is None:
+                return True
+            return actor_matches_allow(actor, database_allow)
+        elif action == "view-table":
+            database, table = resource
+            tables = datasette.metadata("tables", database=database) or {}
+            table_allow = (tables.get(table) or {}).get("allow")
+            if table_allow is None:
+                return True
+            return actor_matches_allow(actor, table_allow)
+        elif action == "view-query":
+            # Check if this query has a "allow" block in metadata
+            database, query_name = resource
+            query = await datasette.get_canned_query(database, query_name, actor)
+            assert query is not None
+            allow = query.get("allow")
+            if allow is None:
+                return True
             return actor_matches_allow(actor, allow)
-    elif action == "view-database":
-        database_allow = datasette.metadata("allow", database=resource)
-        if database_allow is None:
-            return True
-        return actor_matches_allow(actor, database_allow)
-    elif action == "view-table":
-        database, table = resource
-        tables = datasette.metadata("tables", database=database) or {}
-        table_allow = (tables.get(table) or {}).get("allow")
-        if table_allow is None:
-            return True
-        return actor_matches_allow(actor, table_allow)
-    elif action == "view-query":
-        # Check if this query has a "allow" block in metadata
-        database, query_name = resource
-        queries_metadata = datasette.metadata("queries", database=database)
-        assert query_name in queries_metadata
-        if isinstance(queries_metadata[query_name], str):
-            return True
-        allow = queries_metadata[query_name].get("allow")
-        if allow is None:
-            return True
-        return actor_matches_allow(actor, allow)
-    elif action == "execute-sql":
-        # Use allow_sql block from database block, or from top-level
-        database_allow_sql = datasette.metadata("allow_sql", database=resource)
-        if database_allow_sql is None:
-            database_allow_sql = datasette.metadata("allow_sql")
-        if database_allow_sql is None:
-            return True
-        return actor_matches_allow(actor, database_allow_sql)
+        elif action == "execute-sql":
+            # Use allow_sql block from database block, or from top-level
+            database_allow_sql = datasette.metadata("allow_sql", database=resource)
+            if database_allow_sql is None:
+                database_allow_sql = datasette.metadata("allow_sql")
+            if database_allow_sql is None:
+                return True
+            return actor_matches_allow(actor, database_allow_sql)
+
+    return inner
diff --git a/datasette/hookspecs.py b/datasette/hookspecs.py
index 9fceee41..91feb49b 100644
--- a/datasette/hookspecs.py
+++ b/datasette/hookspecs.py
@@ -78,3 +78,8 @@ def actor_from_request(datasette, request):
 @hookspec
 def permission_allowed(datasette, actor, action, resource):
     "Check if actor is allowed to perfom this action - return True, False or None"
+
+
+@hookspec
+def canned_queries(datasette, database, actor):
+    "Return a dictonary of canned query definitions or an awaitable function that returns them"
diff --git a/datasette/views/database.py b/datasette/views/database.py
index 4fab2cfb..ad28fb63 100644
--- a/datasette/views/database.py
+++ b/datasette/views/database.py
@@ -70,7 +70,9 @@ class DatabaseView(DataView):
 
         tables.sort(key=lambda t: (t["hidden"], t["name"]))
         canned_queries = []
-        for query in self.ds.get_canned_queries(database):
+        for query in (
+            await self.ds.get_canned_queries(database, request.actor)
+        ).values():
             visible, private = await check_visibility(
                 self.ds, request.actor, "view-query", (database, query["name"]),
             )
diff --git a/datasette/views/table.py b/datasette/views/table.py
index 91245293..1a55a495 100644
--- a/datasette/views/table.py
+++ b/datasette/views/table.py
@@ -223,7 +223,9 @@ class TableView(RowTableShared):
 
     async def post(self, request, db_name, table_and_format):
         # Handle POST to a canned query
-        canned_query = self.ds.get_canned_query(db_name, table_and_format)
+        canned_query = await self.ds.get_canned_query(
+            db_name, table_and_format, request.actor
+        )
         assert canned_query, "You may only POST to a canned query"
         return await QueryView(self.ds).data(
             request,
@@ -247,7 +249,7 @@ class TableView(RowTableShared):
         _next=None,
         _size=None,
     ):
-        canned_query = self.ds.get_canned_query(database, table)
+        canned_query = await self.ds.get_canned_query(database, table, request.actor)
         if canned_query:
             return await QueryView(self.ds).data(
                 request,
diff --git a/docs/plugins.rst b/docs/plugins.rst
index 113e6b24..8444516c 100644
--- a/docs/plugins.rst
+++ b/docs/plugins.rst
@@ -1031,6 +1031,73 @@ Potential use-cases:
 * Create database tables that a plugin needs
 * Validate the metadata configuration for a plugin on startup, and raise an error if it is invalid
 
+.. _plugin_hook_canned_queries:
+
+canned_queries(datasette, database, actor)
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+``datasette`` - :ref:`internals_datasette`
+    You can use this to access plugin configuration options via ``datasette.plugin_config(your_plugin_name)``, or to execute SQL queries.
+
+``database`` - string
+    The name of the database.
+
+``actor`` - dictionary or None
+    The currently authenticated :ref:`authentication_actor`.
+
+Ues this hook to return a dictionary of additional :ref:`canned query <canned_queries>` definitions for the specified database. The return value should be the same shape as the JSON described in the :ref:`canned query <canned_queries>` documentation.
+
+.. code-block:: python
+
+    from datasette import hookimpl
+
+    @hookimpl
+    def canned_queries(datasette, database):
+        if database == "mydb":
+            return {
+                "my_query": {
+                    "sql": "select * from my_table where id > :min_id"
+                }
+            }
+
+The hook can alternatively return an awaitable function that returns a list. Here's an example that returns queries that have been stored in the ``saved_queries`` database table, if one exists:
+
+.. code-block:: python
+
+    from datasette import hookimpl
+
+    @hookimpl
+    def canned_queries(datasette, database):
+        async def inner():
+            db = datasette.get_database(database)
+            if await db.table_exists("saved_queries"):
+                results = await db.execute("select name, sql from saved_queries")
+                return {result["name"]: {
+                    "sql": result["sql"]
+                } for result in results}
+        return inner
+
+The actor parameter can be used to include the currently authenticated actor in your decision. Here's an example that returns saved queries that were saved by that actor:
+
+.. code-block:: python
+
+    from datasette import hookimpl
+
+    @hookimpl
+    def canned_queries(datasette, database, actor):
+        async def inner():
+            db = datasette.get_database(database)
+            if actor is not None and await db.table_exists("saved_queries"):
+                results = await db.execute(
+                    "select name, sql from saved_queries where actor_id = :id", {
+                        "id": actor["id"]
+                    }
+                )
+                return {result["name"]: {
+                    "sql": result["sql"]
+                } for result in results}
+        return inner
+
 .. _plugin_hook_actor_from_request:
 
 actor_from_request(datasette, request)
diff --git a/tests/fixtures.py b/tests/fixtures.py
index 612bee99..9b28c283 100644
--- a/tests/fixtures.py
+++ b/tests/fixtures.py
@@ -40,6 +40,7 @@ EXPECTED_PLUGINS = [
         "hooks": [
             "actor_from_request",
             "asgi_wrapper",
+            "canned_queries",
             "extra_body_script",
             "extra_css_urls",
             "extra_js_urls",
@@ -61,6 +62,7 @@ EXPECTED_PLUGINS = [
         "hooks": [
             "actor_from_request",
             "asgi_wrapper",
+            "canned_queries",
             "extra_js_urls",
             "extra_template_vars",
             "permission_allowed",
diff --git a/tests/plugins/my_plugin.py b/tests/plugins/my_plugin.py
index a86e3cbf..7ed26908 100644
--- a/tests/plugins/my_plugin.py
+++ b/tests/plugins/my_plugin.py
@@ -193,3 +193,12 @@ def register_routes():
 @hookimpl
 def startup(datasette):
     datasette._startup_hook_fired = True
+
+
+@hookimpl
+def canned_queries(datasette, database, actor):
+    return {
+        "from_hook": "select 1, '{}' as actor_id".format(
+            actor["id"] if actor else "null"
+        )
+    }
diff --git a/tests/plugins/my_plugin_2.py b/tests/plugins/my_plugin_2.py
index f4a082a0..556c8090 100644
--- a/tests/plugins/my_plugin_2.py
+++ b/tests/plugins/my_plugin_2.py
@@ -129,3 +129,17 @@ def startup(datasette):
         datasette._startup_hook_calculation = result.first()[0]
 
     return inner
+
+
+@hookimpl
+def canned_queries(datasette, database):
+    async def inner():
+        return {
+            "from_async_hook": "select {}".format(
+                (
+                    await datasette.get_database(database).execute("select 1 + 1")
+                ).first()[0]
+            )
+        }
+
+    return inner
diff --git a/tests/test_canned_write.py b/tests/test_canned_write.py
index 4257806e..c36baa09 100644
--- a/tests/test_canned_write.py
+++ b/tests/test_canned_write.py
@@ -111,7 +111,13 @@ def test_canned_query_permissions_on_database_page(canned_write_client):
     query_names = [
         q["name"] for q in canned_write_client.get("/data.json").json["queries"]
     ]
-    assert ["add_name", "add_name_specify_id", "update_name"] == query_names
+    assert [
+        "add_name",
+        "add_name_specify_id",
+        "update_name",
+        "from_async_hook",
+        "from_hook",
+    ] == query_names
 
     # With auth shows four
     response = canned_write_client.get(
@@ -124,6 +130,8 @@ def test_canned_query_permissions_on_database_page(canned_write_client):
         {"name": "add_name_specify_id", "private": False},
         {"name": "delete_name", "private": True},
         {"name": "update_name", "private": False},
+        {"name": "from_async_hook", "private": False},
+        {"name": "from_hook", "private": False},
     ] == [
         {"name": q["name"], "private": q["private"]} for q in response.json["queries"]
     ]
diff --git a/tests/test_html.py b/tests/test_html.py
index f9b18daa..7bc935b0 100644
--- a/tests/test_html.py
+++ b/tests/test_html.py
@@ -97,6 +97,8 @@ def test_database_page(app_client):
         ),
         ("/fixtures/pragma_cache_size", "pragma_cache_size"),
         ("/fixtures/neighborhood_search#fragment-goes-here", "Search neighborhoods"),
+        ("/fixtures/from_async_hook", "from_async_hook"),
+        ("/fixtures/from_hook", "from_hook"),
     ] == [(a["href"], a.text) for a in queries_ul.find_all("a")]
 
 
diff --git a/tests/test_plugins.py b/tests/test_plugins.py
index 245c60f7..4f44430e 100644
--- a/tests/test_plugins.py
+++ b/tests/test_plugins.py
@@ -591,3 +591,34 @@ async def test_startup(app_client):
     await app_client.ds.invoke_startup()
     assert app_client.ds._startup_hook_fired
     assert 2 == app_client.ds._startup_hook_calculation
+
+
+def test_canned_queries(app_client):
+    queries = app_client.get("/fixtures.json").json["queries"]
+    queries_by_name = {q["name"]: q for q in queries}
+    assert {
+        "sql": "select 2",
+        "name": "from_async_hook",
+        "private": False,
+    } == queries_by_name["from_async_hook"]
+    assert {
+        "sql": "select 1, 'null' as actor_id",
+        "name": "from_hook",
+        "private": False,
+    } == queries_by_name["from_hook"]
+
+
+def test_canned_queries_non_async(app_client):
+    response = app_client.get("/fixtures/from_hook.json?_shape=array")
+    assert [{"1": 1, "actor_id": "null"}] == response.json
+
+
+def test_canned_queries_async(app_client):
+    response = app_client.get("/fixtures/from_async_hook.json?_shape=array")
+    assert [{"2": 2}] == response.json
+
+
+def test_canned_queries_actor(app_client):
+    assert [{"1": 1, "actor_id": "bot"}] == app_client.get(
+        "/fixtures/from_hook.json?_bot=1&_shape=array"
+    ).json