datasette.allowed_many() method and per-request permission check cache

Merge pull request #2775

datasette/app.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import asyncio
 import contextvars
-from typing import TYPE_CHECKING, Any, Dict, Iterable, List
+from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Sequence
 
 if TYPE_CHECKING:
     from datasette.permissions import Resource
@@ -291,6 +291,15 @@ DEFAULT_NOT_SET = object()
 ResourcesSQL = collections.namedtuple("ResourcesSQL", ("sql", "params"))
 
 
+def _permission_cache_key(actor, action, parent, child):
+    # Key on the full serialized actor so actors differing in any field
+    # (e.g. token restrictions) never share cache entries
+    actor_key = (
+        json.dumps(actor, sort_keys=True, default=repr) if actor is not None else None
+    )
+    return (actor_key, action, parent, child)
+
+
 async def favicon(request, send):
     await asgi_send_file(
         send,
@@ -1817,46 +1826,124 @@ class Datasette:
             # For global actions, resource can be omitted:
             can_debug = await datasette.allowed(action="permissions-debug", actor=actor)
         """
-        from datasette.utils.actions_sql import check_permission_for_resource
+        results = await self.allowed_many(
+            actions=[action], resource=resource, actor=actor
+        )
+        return results[action]
 
-        # For global actions, resource remains None
+    async def allowed_many(
+        self,
+        *,
+        actions: Sequence[str],
+        resource: "Resource" = None,
+        actor: dict | None = None,
+    ) -> dict[str, bool]:
+        """
+        Check several actions against one resource for one actor.
 
-        # Check if this action has also_requires - if so, check that action first
-        action_obj = self.actions.get(action)
-        if action_obj and action_obj.also_requires:
-            # Must have the required action first
-            if not await self.allowed(
-                action=action_obj.also_requires,
-                resource=resource,
+        Resolves every action (plus any also_requires dependencies) with a
+        single internal database query, instead of one or two queries per
+        action. Results are stored in the request-scoped permission cache,
+        so subsequent datasette.allowed() calls for the same checks within
+        the same request are served from the cache.
+
+        Example:
+            from datasette.resources import TableResource
+            results = await datasette.allowed_many(
+                actions=["edit-schema", "drop-table", "insert-row"],
+                resource=TableResource(database="data", table="exercise"),
                 actor=actor,
-            ):
-                return False
+            )
+            # {"edit-schema": True, "drop-table": True, "insert-row": False}
+        """
+        from datasette.utils.actions_sql import check_permissions_for_actions
+        from datasette.permissions import (
+            _permission_check_cache,
+            _skip_permission_checks,
+        )
 
         # For global actions, resource is None
         parent = resource.parent if resource else None
         child = resource.child if resource else None
 
-        result = await check_permission_for_resource(
-            datasette=self,
-            actor=actor,
-            action=action,
-            parent=parent,
-            child=child,
-        )
+        # Expand also_requires dependencies (transitively) so that each
+        # dependency is resolved within the same batch
+        expanded = []
 
-        # Log the permission check for debugging
-        self._permission_checks.append(
-            PermissionCheck(
-                when=datetime.datetime.now(datetime.timezone.utc).isoformat(),
+        def add_action(name):
+            if name in expanded:
+                return
+            action_obj = self.actions.get(name)
+            if action_obj is None:
+                raise ValueError(f"Unknown action: {name}")
+            expanded.append(name)
+            if action_obj.also_requires:
+                add_action(action_obj.also_requires)
+
+        requested = list(dict.fromkeys(actions))
+        for name in requested:
+            add_action(name)
+
+        # Consult the request-scoped cache, unless permission checks are
+        # being skipped (skip-mode verdicts must never be cached)
+        skip = _skip_permission_checks.get()
+        cache = None if skip else _permission_check_cache.get()
+
+        final = {}
+        to_check = []
+        for name in expanded:
+            if cache is not None:
+                key = _permission_cache_key(actor, name, parent, child)
+                if key in cache:
+                    final[name] = cache[key]
+                    continue
+            to_check.append(name)
+
+        raw = {}
+        if to_check:
+            raw = await check_permissions_for_actions(
+                datasette=self,
                 actor=actor,
-                action=action,
+                actions=to_check,
                 parent=parent,
                 child=child,
-                result=result,
             )
-        )
 
-        return result
+        def resolve(name):
+            # final verdict = own rules AND verdict of also_requires chain
+            if name in final:
+                return final[name]
+            result = raw[name]
+            action_obj = self.actions.get(name)
+            if result and action_obj.also_requires:
+                result = resolve(action_obj.also_requires)
+            final[name] = result
+            return result
+
+        for name in expanded:
+            resolve(name)
+
+        # Cache the freshly computed checks
+        if cache is not None:
+            for name in to_check:
+                cache[_permission_cache_key(actor, name, parent, child)] = final[name]
+
+        # Log every check (including cache hits) for the debug page,
+        # dependencies before the actions that required them
+        when = datetime.datetime.now(datetime.timezone.utc).isoformat()
+        for name in reversed(expanded):
+            self._permission_checks.append(
+                PermissionCheck(
+                    when=when,
+                    actor=actor,
+                    action=name,
+                    parent=parent,
+                    child=child,
+                    result=final[name],
+                )
+            )
+
+        return {name: final[name] for name in requested}
 
     async def ensure_permission(
         self,
@@ -2612,7 +2699,16 @@ class DatasetteRouter:
         if raw_path:
             path = raw_path.decode("ascii")
         path = path.partition("?")[0]
-        return await self.route_path(scope, receive, send, path)
+        # Give each request a fresh permission check cache, so repeated
+        # datasette.allowed() checks within the request are memoized but
+        # results never persist beyond it
+        from datasette.permissions import _permission_check_cache
+
+        cache_token = _permission_check_cache.set({})
+        try:
+            return await self.route_path(scope, receive, send, path)
+        finally:
+            _permission_check_cache.reset(cache_token)
 
     async def route_path(self, scope, receive, send, path):
         # Strip off base_url if present before routing

datasette/permissions.py

@@ -8,6 +8,14 @@ _skip_permission_checks = contextvars.ContextVar(
     "skip_permission_checks", default=False
 )
 
+# Request-scoped cache of permission check results. The ASGI router sets
+# this to a fresh dict at the start of each request, so cached verdicts
+# never outlive a request or leak between actors. Keys are
+# (actor_json, action, parent, child) tuples, values are booleans.
+_permission_check_cache: contextvars.ContextVar[dict | None] = contextvars.ContextVar(
+    "permission_check_cache", default=None
+)
+
 
 class SkipPermissions:
     """Context manager to temporarily skip permission checks.

datasette/utils/actions_sql.py

@@ -21,6 +21,8 @@ The core pattern is:
 - Across levels, child beats parent beats global
 """
 
+import asyncio
+import re
 from typing import TYPE_CHECKING
 
 from datasette.utils.permissions import gather_permission_sql_from_hooks
@@ -495,6 +497,153 @@ async def build_permission_rules_sql(
     return rules_union, all_params, restriction_sqls
 
 
+async def check_permissions_for_actions(
+    *,
+    datasette: "Datasette",
+    actor: dict | None,
+    actions: list[str],
+    parent: str | None,
+    child: str | None,
+) -> dict[str, bool]:
+    """
+    Check several actions for one actor and resource in a single query.
+
+    Args:
+        datasette: The Datasette instance
+        actor: The actor dict (or None)
+        actions: List of action names to check
+        parent: The parent resource identifier (e.g., database name, or None)
+        child: The child resource identifier (e.g., table name, or None)
+
+    Returns:
+        Dict mapping each action name to True (allowed) or False (denied)
+
+    Each action contributes its own tagged block of permission rules
+    (gathered from the permission_resources_sql hook, with parameters
+    namespaced per action to avoid collisions) plus an optional
+    restriction allowlist CTE. One internal database query resolves
+    the winning rule per action using the same specificity-then-deny
+    ordering as the rest of the permission system.
+
+    Note: this resolves each action independently - also_requires
+    dependencies are handled by the caller (Datasette.allowed_many).
+    """
+    from datasette.utils.permissions import SKIP_PERMISSION_CHECKS
+
+    for action in actions:
+        if not datasette.actions.get(action):
+            raise ValueError(f"Unknown action: {action}")
+
+    # Dedupe while preserving order
+    unique_actions = list(dict.fromkeys(actions))
+    if not unique_actions:
+        return {}
+
+    # Gather hook results for each action concurrently - hooks within a
+    # single action still run sequentially, preserving existing semantics
+    gathered = await asyncio.gather(
+        *(
+            gather_permission_sql_from_hooks(
+                datasette=datasette, actor=actor, action=action
+            )
+            for action in unique_actions
+        )
+    )
+
+    if any(result is SKIP_PERMISSION_CHECKS for result in gathered):
+        return {action: True for action in unique_actions}
+
+    params = {"_check_parent": parent, "_check_child": child}
+    ctes = []
+    result_rows = []
+    verdicts = {}
+
+    for i, (action, permission_sqls) in enumerate(zip(unique_actions, gathered)):
+        prefix = f"a{i}_"
+        rule_parts = []
+        restriction_parts = []
+
+        for permission_sql in permission_sqls:
+            sql = permission_sql.sql
+            restriction_sql = permission_sql.restriction_sql
+            # Namespace this block's params so identical names used for
+            # different actions cannot collide
+            for key in permission_sql.params or {}:
+                new_key = prefix + key
+                params[new_key] = permission_sql.params[key]
+                pattern = re.compile(":" + re.escape(key) + r"(?![A-Za-z0-9_])")
+                if sql:
+                    sql = pattern.sub(":" + new_key, sql)
+                if restriction_sql:
+                    restriction_sql = pattern.sub(":" + new_key, restriction_sql)
+
+            if restriction_sql:
+                restriction_parts.append(restriction_sql)
+
+            # Skip plugins that only provide restriction_sql (no permission rules)
+            if sql is None:
+                continue
+            rule_parts.append(
+                f"SELECT parent, child, allow, reason, '{permission_sql.source}' AS source_plugin FROM (\n{sql}\n)"
+            )
+
+        if not rule_parts:
+            # No rules from any plugin - default deny. Restrictions can
+            # only restrict, never grant, so no SQL is needed at all
+            verdicts[action] = False
+            continue
+        ctes.append(f"a{i}_rules AS (\n" + "\nUNION ALL\n".join(rule_parts) + "\n)")
+
+        # Winning rule for this action: most specific depth first, then
+        # deny-beats-allow, then source_plugin as a stable tie-break
+        verdict_sql = f"""COALESCE((
+  SELECT allow FROM (
+    SELECT allow, source_plugin,
+      CASE
+        WHEN child IS NOT NULL THEN 2
+        WHEN parent IS NOT NULL THEN 1
+        ELSE 0
+      END AS depth
+    FROM a{i}_rules
+    WHERE (parent IS NULL OR parent = :_check_parent)
+      AND (child IS NULL OR child = :_check_child)
+    ORDER BY
+      depth DESC,
+      CASE WHEN allow = 0 THEN 0 ELSE 1 END,
+      source_plugin
+    LIMIT 1
+  )
+), 0)"""
+
+        if restriction_parts:
+            # Database-level restrictions (parent, NULL) match all children
+            restriction_intersect = "\nINTERSECT\n".join(
+                f"SELECT * FROM ({sql})" for sql in restriction_parts
+            )
+            ctes.append(f"a{i}_restriction AS (\n{restriction_intersect}\n)")
+            verdict_sql = f"""({verdict_sql}) AND EXISTS (
+  SELECT 1 FROM a{i}_restriction r
+  WHERE (r.parent = :_check_parent OR r.parent IS NULL)
+    AND (r.child = :_check_child OR r.child IS NULL)
+)"""
+
+        result_rows.append(f"({i}, ({verdict_sql}))")
+
+    if result_rows:
+        ctes.append(
+            "results(action_idx, is_allowed) AS (VALUES\n"
+            + ",\n".join(result_rows)
+            + "\n)"
+        )
+        query = (
+            "WITH\n" + ",\n".join(ctes) + "\nSELECT action_idx, is_allowed FROM results"
+        )
+        result = await datasette.get_internal_database().execute(query, params)
+        for row in result.rows:
+            verdicts[unique_actions[row[0]]] = bool(row[1])
+    return verdicts
+
+
 async def check_permission_for_resource(
     *,
     datasette: "Datasette",
@@ -515,77 +664,12 @@ async def check_permission_for_resource(
 
     Returns:
         True if the actor is allowed, False otherwise
-
-    This builds the cascading permission query and checks if the specific
-    resource is in the allowed set.
     """
-    rules_union, all_params, restriction_sqls = await build_permission_rules_sql(
-        datasette, actor, action
+    results = await check_permissions_for_actions(
+        datasette=datasette,
+        actor=actor,
+        actions=[action],
+        parent=parent,
+        child=child,
     )
-
-    # If no rules (empty SQL), default deny
-    if not rules_union:
-        return False
-
-    # Add parameters for the resource we're checking
-    all_params["_check_parent"] = parent
-    all_params["_check_child"] = child
-
-    # If there are restriction filters, check if the resource passes them first
-    if restriction_sqls:
-        # Check if resource is in restriction allowlist
-        # Database-level restrictions (parent, NULL) should match all children (parent, *)
-        # Wrap each restriction_sql in a subquery to avoid operator precedence issues
-        restriction_check = "\nINTERSECT\n".join(
-            f"SELECT * FROM ({sql})" for sql in restriction_sqls
-        )
-        restriction_query = f"""
-WITH restriction_list AS (
-    {restriction_check}
-)
-SELECT EXISTS (
-    SELECT 1 FROM restriction_list
-    WHERE (parent = :_check_parent OR parent IS NULL)
-      AND (child = :_check_child OR child IS NULL)
-) AS in_allowlist
-"""
-        result = await datasette.get_internal_database().execute(
-            restriction_query, all_params
-        )
-        if result.rows and not result.rows[0][0]:
-            # Resource not in restriction allowlist - deny
-            return False
-
-    query = f"""
-WITH
-all_rules AS (
-  {rules_union}
-),
-matched_rules AS (
-  SELECT ar.*,
-    CASE
-      WHEN ar.child IS NOT NULL THEN 2  -- child-level (most specific)
-      WHEN ar.parent IS NOT NULL THEN 1  -- parent-level
-      ELSE 0                             -- root/global
-    END AS depth
-  FROM all_rules ar
-  WHERE (ar.parent IS NULL OR ar.parent = :_check_parent)
-    AND (ar.child IS NULL OR ar.child = :_check_child)
-),
-winner AS (
-  SELECT *
-  FROM matched_rules
-  ORDER BY
-    depth DESC,                          -- specificity first (higher depth wins)
-    CASE WHEN allow=0 THEN 0 ELSE 1 END, -- then deny over allow
-    source_plugin                        -- stable tie-break
-  LIMIT 1
-)
-SELECT COALESCE((SELECT allow FROM winner), 0) AS is_allowed
-"""
-
-    # Execute the query against the internal database
-    result = await datasette.get_internal_database().execute(query, all_params)
-    if result.rows:
-        return bool(result.rows[0][0])
-    return False
+    return results[action]

datasette/views/database.py

@@ -118,6 +118,19 @@ class DatabaseView(View):
         )
 
         async def database_actions():
+            # Resolve the registered database-level actions for this
+            # database in one batched query, seeding the request permission
+            # cache so that allowed() calls made inside the plugin hooks
+            # below are served from the cache
+            await datasette.allowed_many(
+                actions=[
+                    name
+                    for name, action in datasette.actions.items()
+                    if action.resource_class is DatabaseResource
+                ],
+                resource=DatabaseResource(database),
+                actor=request.actor,
+            )
             links = []
             for hook in pm.hook.database_actions(
                 datasette=datasette,

datasette/views/table_extras.py

@@ -4,7 +4,7 @@ from dataclasses import dataclass
 from datasette.database import QueryInterrupted
 from datasette.extras import Extra, ExtraExample, ExtraRegistry, ExtraScope, Provider
 from datasette.plugins import pm
-from datasette.resources import TableResource
+from datasette.resources import DatabaseResource, TableResource
 from datasette.utils import (
     await_me_maybe,
     call_with_supported_arguments,
@@ -361,6 +361,30 @@ class ActionsExtra(Extra):
             else:
                 kwargs["table"] = context.table_name
                 method = pm.hook.table_actions
+                # Resolve the registered table-level actions for this table
+                # and the database-level actions for its database in two
+                # batched queries, seeding the request permission cache so
+                # that allowed() calls made inside the plugin hooks below
+                # are served from the cache
+                datasette = context.datasette
+                await datasette.allowed_many(
+                    actions=[
+                        name
+                        for name, action in datasette.actions.items()
+                        if action.resource_class is TableResource
+                    ],
+                    resource=TableResource(context.database_name, context.table_name),
+                    actor=context.request.actor,
+                )
+                await datasette.allowed_many(
+                    actions=[
+                        name
+                        for name, action in datasette.actions.items()
+                        if action.resource_class is DatabaseResource
+                    ],
+                    resource=DatabaseResource(context.database_name),
+                    actor=context.request.actor,
+                )
             for hook in method(**kwargs):
                 extra_links = await await_me_maybe(hook)
                 if extra_links:

docs/internals.rst

@@ -512,6 +512,43 @@ Example usage:
 
 The method returns ``True`` if the permission is granted, ``False`` if denied.
 
+Results are cached for the duration of the current request, so checking the same ``(actor, action, resource)`` combination twice within one request only does the underlying permission resolution work once.
+
+.. _datasette_allowed_many:
+
+await .allowed_many(\*, actions, resource, actor=None)
+------------------------------------------------------
+
+``actions`` - list of strings
+    The names of the actions to permission check.
+
+``resource`` - Resource object
+    A Resource object representing the database, table, or other resource that each action is checked against. Omit for global actions.
+
+``actor`` - dictionary, optional
+    The authenticated actor. This is usually ``request.actor``. Defaults to ``None`` for unauthenticated requests.
+
+Checks several actions against the same resource for the same actor, returning a dictionary mapping each action name to ``True`` or ``False``. The whole batch - including any actions pulled in through ``also_requires`` dependencies - is resolved with a single SQL query against the internal database, so this is much faster than calling :ref:`datasette.allowed() <datasette_allowed>` once per action.
+
+Example usage:
+
+.. code-block:: python
+
+    from datasette.resources import TableResource
+
+    results = await datasette.allowed_many(
+        actions=["insert-row", "delete-row", "drop-table"],
+        resource=TableResource(
+            database="fixtures", table="facetable"
+        ),
+        actor=request.actor,
+    )
+    # {"insert-row": True, "delete-row": True, "drop-table": False}
+
+Each result is stored in the per-request permission check cache, so subsequent ``datasette.allowed()`` calls for the same checks within the same request are served from that cache. Datasette uses this before running the ``table_actions`` and ``database_actions`` plugin hooks: it resolves every registered table-level action against the current table and every database-level action against its database first, which means ``allowed()`` calls made by those plugin hooks are usually served from the cache instead of triggering additional queries.
+
+Actions for which no plugin provides any permission rules are resolved to ``False`` directly, without being included in the SQL query at all.
+
 .. _datasette_allowed_resources:
 
 await .allowed_resources(action, actor=None, \*, parent=None, include_is_private=False, include_reasons=False, limit=100, next=None)

tests/conftest.py

@@ -146,6 +146,7 @@ def restore_working_directory(tmpdir, request):
 @pytest.fixture(scope="session", autouse=True)
 def check_actions_are_documented():
     from datasette.plugins import pm
+    from datasette.default_actions import register_actions as default_register_actions
 
     content = (
         pathlib.Path(__file__).parent.parent / "docs" / "authentication.rst"
@@ -154,6 +155,9 @@ def check_actions_are_documented():
     documented_actions = set(permissions_re.findall(content)).union(
         UNDOCUMENTED_PERMISSIONS
     )
+    # Only Datasette core actions need to be documented - actions registered
+    # by (test) plugins are checked for registration but not documentation
+    core_actions = {action.name for action in default_register_actions()}
 
     def before(hook_name, hook_impls, kwargs):
         if hook_name == "permission_resources_sql":
@@ -165,9 +169,10 @@ def check_actions_are_documented():
                 + " (or maybe a test forgot to do await ds.invoke_startup())"
             )
             action = kwargs.get("action").replace("-", "_")
-            assert (
-                action in documented_actions
-            ), "Undocumented permission action: {}".format(action)
+            if kwargs["action"] in core_actions:
+                assert (
+                    action in documented_actions
+                ), "Undocumented permission action: {}".format(action)
 
     pm.add_hookcall_monitoring(
         before=before, after=lambda outcome, hook_name, hook_impls, kwargs: None

tests/test_allowed_many.py

@@ -0,0 +1,707 @@
+"""
+Tests for request-scoped permission check memoization and the
+datasette.allowed_many() batch permission API.
+
+Layer 1: per-request cache consulted by datasette.allowed()
+Layer 2: allowed_many() resolves multiple actions in one internal-DB query
+Layer 3: table/database views precompute all registered actions before
+         invoking table_actions/database_actions plugin hooks
+"""
+
+import pytest
+import pytest_asyncio
+from datasette.app import Datasette
+from datasette.permissions import (
+    Action,
+    PermissionSQL,
+    SkipPermissions,
+    _permission_check_cache,
+)
+from datasette.resources import DatabaseResource, TableResource
+from datasette import hookimpl
+
+
+class CountingRulesPlugin:
+    """Counts permission_resources_sql gathers and grants rules for alice."""
+
+    def __init__(self):
+        self.calls = []
+
+    @hookimpl
+    def permission_resources_sql(self, datasette, actor, action):
+        actor_id = actor.get("id") if actor else None
+        self.calls.append((actor_id, action))
+        if actor_id == "alice":
+            return PermissionSQL(
+                sql="SELECT NULL AS parent, NULL AS child, 1 AS allow, 'alice allowed' AS reason"
+            )
+        return None
+
+    def count(self, actor_id=None, action=None):
+        return len(
+            [
+                (a, c)
+                for a, c in self.calls
+                if (actor_id is None or a == actor_id)
+                and (action is None or c == action)
+            ]
+        )
+
+
+@pytest_asyncio.fixture
+async def ds():
+    ds = Datasette()
+    await ds.invoke_startup()
+    db = ds.add_memory_database("analytics")
+    await db.execute_write("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY)")
+    await db.execute_write("CREATE TABLE IF NOT EXISTS events (id INTEGER PRIMARY KEY)")
+    await ds._refresh_schemas()
+    return ds
+
+
+@pytest_asyncio.fixture
+async def counting_ds(ds):
+    plugin = CountingRulesPlugin()
+    ds.pm.register(plugin, name="counting")
+    try:
+        yield ds, plugin
+    finally:
+        ds.pm.unregister(name="counting")
+
+
+# ----------------------------------------------------------------------
+# Layer 1: request-scoped memoization
+# ----------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_allowed_memoized_when_cache_active(counting_ds):
+    ds, plugin = counting_ds
+    resource = TableResource("analytics", "users")
+    token = _permission_check_cache.set({})
+    try:
+        first = await ds.allowed(
+            action="view-table", resource=resource, actor={"id": "alice"}
+        )
+        gathers_after_first = plugin.count(actor_id="alice", action="view-table")
+        assert gathers_after_first > 0
+        second = await ds.allowed(
+            action="view-table", resource=resource, actor={"id": "alice"}
+        )
+        assert first is True
+        assert second is True
+        # The second identical check must not gather hooks again
+        assert plugin.count(actor_id="alice", action="view-table") == (
+            gathers_after_first
+        )
+    finally:
+        _permission_check_cache.reset(token)
+
+
+@pytest.mark.asyncio
+async def test_allowed_not_memoized_without_cache(counting_ds):
+    ds, plugin = counting_ds
+    resource = TableResource("analytics", "users")
+    assert _permission_check_cache.get() is None
+    await ds.allowed(action="view-table", resource=resource, actor={"id": "alice"})
+    first_count = plugin.count(actor_id="alice", action="view-table")
+    await ds.allowed(action="view-table", resource=resource, actor={"id": "alice"})
+    # No request cache active - hooks gathered again
+    assert plugin.count(actor_id="alice", action="view-table") == first_count * 2
+
+
+@pytest.mark.asyncio
+async def test_cache_keyed_on_full_actor_identity(counting_ds):
+    """Interleaved checks for different actors never share cache entries."""
+    # Uses drop-table because default permissions deny it to non-root actors
+    ds, plugin = counting_ds
+    resource = TableResource("analytics", "users")
+    token = _permission_check_cache.set({})
+    try:
+        assert (
+            await ds.allowed(
+                action="drop-table", resource=resource, actor={"id": "alice"}
+            )
+            is True
+        )
+        assert (
+            await ds.allowed(
+                action="drop-table", resource=resource, actor={"id": "bob"}
+            )
+            is False
+        )
+        # Repeat interleaved - cached results must stay correct per actor
+        assert (
+            await ds.allowed(
+                action="drop-table", resource=resource, actor={"id": "alice"}
+            )
+            is True
+        )
+        assert (
+            await ds.allowed(
+                action="drop-table", resource=resource, actor={"id": "bob"}
+            )
+            is False
+        )
+        # Actors differing in fields beyond id must not collide either
+        assert (
+            await ds.allowed(
+                action="drop-table",
+                resource=resource,
+                actor={"id": "alice", "_r": {"a": []}},
+            )
+            is False
+        )
+    finally:
+        _permission_check_cache.reset(token)
+
+
+@pytest.mark.asyncio
+async def test_cache_keyed_on_resource(counting_ds):
+    ds, plugin = counting_ds
+    token = _permission_check_cache.set({})
+    try:
+        await ds.allowed(
+            action="view-table",
+            resource=TableResource("analytics", "users"),
+            actor={"id": "alice"},
+        )
+        count = plugin.count(actor_id="alice", action="view-table")
+        # Different resource - must not be served from cache
+        await ds.allowed(
+            action="view-table",
+            resource=TableResource("analytics", "events"),
+            actor={"id": "alice"},
+        )
+        assert plugin.count(actor_id="alice", action="view-table") == count * 2
+    finally:
+        _permission_check_cache.reset(token)
+
+
+@pytest.mark.asyncio
+async def test_skip_permission_checks_bypasses_cache(counting_ds):
+    ds, plugin = counting_ds
+    resource = TableResource("analytics", "users")
+    token = _permission_check_cache.set({})
+    try:
+        with SkipPermissions():
+            assert (
+                await ds.allowed(
+                    action="drop-table", resource=resource, actor={"id": "bob"}
+                )
+                is True
+            )
+        # The skip-mode True must not have been cached
+        assert (
+            await ds.allowed(
+                action="drop-table", resource=resource, actor={"id": "bob"}
+            )
+            is False
+        )
+    finally:
+        _permission_check_cache.reset(token)
+
+
+# ----------------------------------------------------------------------
+# Layer 2: allowed_many()
+# ----------------------------------------------------------------------
+
+
+class MatrixRulesPlugin:
+    """Different rules per action for actor carol, to exercise resolution."""
+
+    @hookimpl
+    def permission_resources_sql(self, datasette, actor, action):
+        if not actor or actor.get("id") != "carol":
+            return None
+        if action == "view-table":
+            return PermissionSQL(sql="""
+                SELECT NULL AS parent, NULL AS child, 1 AS allow, 'global allow' AS reason
+                UNION ALL
+                SELECT 'analytics' AS parent, 'sensitive' AS child, 0 AS allow, 'deny sensitive' AS reason
+                """)
+        if action == "insert-row":
+            return PermissionSQL(
+                sql="SELECT 'analytics' AS parent, NULL AS child, 1 AS allow, 'analytics writes' AS reason"
+            )
+        # Everything else: no opinion (implicit deny unless defaults allow)
+        return None
+
+
+@pytest.mark.asyncio
+async def test_allowed_many_basic(ds):
+    plugin = MatrixRulesPlugin()
+    ds.pm.register(plugin, name="matrix")
+    try:
+        results = await ds.allowed_many(
+            actions=["view-table", "insert-row", "drop-table"],
+            resource=TableResource("analytics", "users"),
+            actor={"id": "carol"},
+        )
+        assert results == {
+            "view-table": True,
+            "insert-row": True,
+            "drop-table": False,
+        }
+        # Child-level deny beats global allow
+        sensitive = await ds.allowed_many(
+            actions=["view-table"],
+            resource=TableResource("analytics", "sensitive"),
+            actor={"id": "carol"},
+        )
+        assert sensitive == {"view-table": False}
+    finally:
+        ds.pm.unregister(name="matrix")
+
+
+@pytest.mark.asyncio
+async def test_allowed_many_matches_allowed(ds):
+    """Every action resolved by allowed_many() must match allowed()."""
+    plugin = MatrixRulesPlugin()
+    ds.pm.register(plugin, name="matrix")
+    try:
+        all_actions = list(ds.actions)
+        for resource in (
+            TableResource("analytics", "users"),
+            TableResource("analytics", "sensitive"),
+            DatabaseResource("analytics"),
+        ):
+            batched = await ds.allowed_many(
+                actions=all_actions, resource=resource, actor={"id": "carol"}
+            )
+            assert set(batched) == set(all_actions)
+            for action in all_actions:
+                individual = await ds.allowed(
+                    action=action, resource=resource, actor={"id": "carol"}
+                )
+                assert (
+                    batched[action] == individual
+                ), f"Mismatch for {action} on {resource}"
+    finally:
+        ds.pm.unregister(name="matrix")
+
+
+@pytest.mark.asyncio
+async def test_allowed_many_unknown_action_raises(ds):
+    with pytest.raises(ValueError, match="Unknown action"):
+        await ds.allowed_many(
+            actions=["view-table", "no-such-action"],
+            resource=TableResource("analytics", "users"),
+            actor=None,
+        )
+
+
+@pytest.mark.asyncio
+async def test_allowed_many_empty_actions(ds):
+    assert (
+        await ds.allowed_many(
+            actions=[], resource=TableResource("analytics", "users"), actor=None
+        )
+        == {}
+    )
+
+
+class AlsoRequiresRulesPlugin:
+    """dave: store-query allowed but execute-sql explicitly denied.
+    erin: store-query allowed (execute-sql stays default-allowed)."""
+
+    @hookimpl
+    def permission_resources_sql(self, datasette, actor, action):
+        actor_id = actor.get("id") if actor else None
+        if actor_id == "dave":
+            if action == "store-query":
+                return PermissionSQL(
+                    sql="SELECT NULL AS parent, NULL AS child, 1 AS allow, 'dave can store' AS reason"
+                )
+            if action == "execute-sql":
+                return PermissionSQL(
+                    sql="SELECT NULL AS parent, NULL AS child, 0 AS allow, 'dave no sql' AS reason"
+                )
+        if actor_id == "erin" and action == "store-query":
+            return PermissionSQL(
+                sql="SELECT NULL AS parent, NULL AS child, 1 AS allow, 'erin can store' AS reason"
+            )
+        return None
+
+
+@pytest.mark.asyncio
+async def test_allowed_many_also_requires(ds):
+    # store-query also_requires execute-sql, which also_requires view-database
+    plugin = AlsoRequiresRulesPlugin()
+    ds.pm.register(plugin, name="also_requires")
+    try:
+        resource = DatabaseResource("analytics")
+        dave = await ds.allowed_many(
+            actions=["store-query", "execute-sql", "view-database"],
+            resource=resource,
+            actor={"id": "dave"},
+        )
+        # execute-sql denied, so store-query must be denied too
+        assert dave == {
+            "store-query": False,
+            "execute-sql": False,
+            "view-database": True,
+        }
+        erin = await ds.allowed_many(
+            actions=["store-query"], resource=resource, actor={"id": "erin"}
+        )
+        assert erin == {"store-query": True}
+        # Must match the single-check path
+        assert (
+            await ds.allowed(
+                action="store-query", resource=resource, actor={"id": "dave"}
+            )
+            is False
+        )
+        assert (
+            await ds.allowed(
+                action="store-query", resource=resource, actor={"id": "erin"}
+            )
+            is True
+        )
+    finally:
+        ds.pm.unregister(name="also_requires")
+
+
+@pytest.mark.asyncio
+async def test_allowed_many_respects_restrictions(ds):
+    """Token-style _r restrictions are enforced within the batch."""
+    actor = {"id": "root", "_r": {"d": {"analytics": ["vt"]}}}
+    results = await ds.allowed_many(
+        actions=["view-table", "drop-table"],
+        resource=TableResource("analytics", "users"),
+        actor=actor,
+    )
+    # root could normally do both, but the token only allows view-table
+    # on the analytics database
+    assert results == {"view-table": True, "drop-table": False}
+    other_db = await ds.allowed_many(
+        actions=["view-table"],
+        resource=TableResource("production", "stuff"),
+        actor=actor,
+    )
+    assert other_db == {"view-table": False}
+    # Equivalence with allowed()
+    assert (
+        await ds.allowed(
+            action="view-table",
+            resource=TableResource("analytics", "users"),
+            actor=actor,
+        )
+        is True
+    )
+    assert (
+        await ds.allowed(
+            action="drop-table",
+            resource=TableResource("analytics", "users"),
+            actor=actor,
+        )
+        is False
+    )
+
+
+class ParamCollisionPlugin:
+    """Same parameter name with a different value for every action."""
+
+    @hookimpl
+    def permission_resources_sql(self, datasette, actor, action):
+        if not actor or actor.get("id") != "paula":
+            return None
+        flag = 1 if action in ("drop-table", "insert-row") else 0
+        return PermissionSQL(
+            sql="SELECT NULL AS parent, NULL AS child, :flag AS allow, 'flagged' AS reason",
+            params={"flag": flag},
+        )
+
+
+@pytest.mark.asyncio
+async def test_allowed_many_namespaces_params_across_actions(ds):
+    """40+ actions whose rules use identical param names must not collide."""
+    plugin = ParamCollisionPlugin()
+    ds.pm.register(plugin, name="collision")
+    try:
+        all_actions = list(ds.actions)
+        assert len(all_actions) >= 15
+        resource = TableResource("analytics", "users")
+        results = await ds.allowed_many(
+            actions=all_actions, resource=resource, actor={"id": "paula"}
+        )
+        # Spot-check: only the flagged actions resolve True
+        assert results["drop-table"] is True
+        assert results["create-table"] is False
+        # Full equivalence against single checks
+        for action in all_actions:
+            assert results[action] == await ds.allowed(
+                action=action, resource=resource, actor={"id": "paula"}
+            ), f"Mismatch for {action}"
+    finally:
+        ds.pm.unregister(name="collision")
+
+
+@pytest.mark.asyncio
+async def test_allowed_many_single_internal_db_query(ds):
+    internal_db = ds.get_internal_database()
+    calls = []
+    original_execute = internal_db.execute
+
+    async def counting_execute(sql, params=None, **kwargs):
+        calls.append(sql)
+        return await original_execute(sql, params, **kwargs)
+
+    internal_db.execute = counting_execute
+    try:
+        results = await ds.allowed_many(
+            actions=["view-table", "insert-row", "delete-row", "drop-table"],
+            resource=TableResource("analytics", "users"),
+            actor={"id": "root", "_r": {"d": {"analytics": ["vt"]}}},
+        )
+        assert len(results) == 4
+        assert len(calls) == 1
+    finally:
+        internal_db.execute = original_execute
+
+
+@pytest.mark.asyncio
+async def test_allowed_many_no_query_when_no_rules(ds):
+    """Actions with no rules from any plugin are denied without SQL.
+
+    Restrictions can only restrict, never grant, so an action with no
+    rule rows is always False - it should not contribute to the query,
+    and if no action has rules there should be no query at all."""
+    internal_db = ds.get_internal_database()
+    calls = []
+    original_execute = internal_db.execute
+
+    async def counting_execute(sql, params=None, **kwargs):
+        calls.append(sql)
+        return await original_execute(sql, params, **kwargs)
+
+    internal_db.execute = counting_execute
+    try:
+        # bob gets no rules at all for these write actions
+        results = await ds.allowed_many(
+            actions=["drop-table", "delete-row"],
+            resource=TableResource("analytics", "users"),
+            actor={"id": "bob"},
+        )
+        assert results == {"drop-table": False, "delete-row": False}
+        assert len(calls) == 0
+        # A mixed batch still needs exactly one query
+        calls.clear()
+        results = await ds.allowed_many(
+            actions=["view-table", "drop-table"],
+            resource=TableResource("analytics", "users"),
+            actor={"id": "bob"},
+        )
+        assert results == {"view-table": True, "drop-table": False}
+        assert len(calls) == 1
+    finally:
+        internal_db.execute = original_execute
+
+
+@pytest.mark.asyncio
+async def test_allowed_many_global_actions_without_resource(ds):
+    results = await ds.allowed_many(
+        actions=["view-instance", "permissions-debug"],
+        actor={"id": "root"},
+    )
+    assert results["view-instance"] is True
+    # Equivalence with single checks for global actions
+    for action in ("view-instance", "permissions-debug"):
+        assert results[action] == await ds.allowed(action=action, actor={"id": "root"})
+    anon = await ds.allowed_many(actions=["permissions-debug"], actor=None)
+    assert anon == {"permissions-debug": False}
+
+
+@pytest.mark.asyncio
+async def test_allowed_many_seeds_request_cache(counting_ds):
+    ds, plugin = counting_ds
+    resource = TableResource("analytics", "users")
+    actions = ["view-table", "insert-row", "drop-table"]
+    token = _permission_check_cache.set({})
+    try:
+        await ds.allowed_many(actions=actions, resource=resource, actor={"id": "alice"})
+        gathers = plugin.count(actor_id="alice")
+        assert gathers > 0
+        for action in actions:
+            await ds.allowed(action=action, resource=resource, actor={"id": "alice"})
+        # Every allowed() call must have been served from the seeded cache
+        assert plugin.count(actor_id="alice") == gathers
+    finally:
+        _permission_check_cache.reset(token)
+
+
+@pytest.mark.asyncio
+async def test_allowed_many_skip_permission_checks(ds):
+    with SkipPermissions():
+        results = await ds.allowed_many(
+            actions=["view-table", "drop-table"],
+            resource=TableResource("analytics", "users"),
+            actor=None,
+        )
+    assert results == {"view-table": True, "drop-table": True}
+
+
+class ManyActionsPlugin:
+    """Registers enough actions to exceed SQLite's compound SELECT limit."""
+
+    def __init__(self, count):
+        self.action_names = [f"bulk-action-{i}" for i in range(count)]
+        self.action_names_set = set(self.action_names)
+
+    @hookimpl
+    def register_actions(self, datasette):
+        return [
+            Action(name=name, abbr=None, description="Bulk test action")
+            for name in self.action_names
+        ]
+
+    @hookimpl
+    def permission_resources_sql(self, datasette, actor, action):
+        if action in self.action_names_set:
+            return PermissionSQL(
+                sql="SELECT NULL AS parent, NULL AS child, 1 AS allow, 'bulk allow' AS reason",
+                params={},
+            )
+
+
+@pytest.mark.asyncio
+async def test_allowed_many_more_than_sqlite_compound_select_limit():
+    plugin = ManyActionsPlugin(600)
+    ds = Datasette()
+    ds.pm.register(plugin, name="many_actions")
+    try:
+        await ds.invoke_startup()
+        results = await ds.allowed_many(actions=plugin.action_names, actor=None)
+        assert len(results) == 600
+        assert all(results.values())
+    finally:
+        ds.pm.unregister(name="many_actions")
+
+
+# ----------------------------------------------------------------------
+# Layer 3: precompute before table_actions / database_actions hooks
+# ----------------------------------------------------------------------
+
+
+class ActionHooksPlugin:
+    """Plugin hooks that make allowed() checks, like real action plugins do."""
+
+    @hookimpl
+    def table_actions(self, datasette, actor, database, table):
+        async def inner():
+            links = []
+            if await datasette.allowed(
+                action="drop-table",
+                resource=TableResource(database, table),
+                actor=actor,
+            ):
+                links.append(
+                    {"href": "/drop", "label": "Drop this table (test-plugin)"}
+                )
+            if await datasette.allowed(
+                action="create-table",
+                resource=DatabaseResource(database),
+                actor=actor,
+            ):
+                links.append(
+                    {"href": "/create", "label": "Create a table (test-plugin)"}
+                )
+            return links
+
+        return inner
+
+    @hookimpl
+    def database_actions(self, datasette, actor, database):
+        async def inner():
+            if await datasette.allowed(
+                action="create-table",
+                resource=DatabaseResource(database),
+                actor=actor,
+            ):
+                return [{"href": "/create", "label": "Create a table (test-plugin)"}]
+            return []
+
+        return inner
+
+
+@pytest_asyncio.fixture
+async def spying_ds(ds, monkeypatch):
+    """ds with the ActionHooksPlugin plus a spy recording every batch of
+    actions sent to check_permissions_for_actions."""
+    from datasette.utils import actions_sql
+
+    plugin = ActionHooksPlugin()
+    ds.pm.register(plugin, name="action_hooks")
+    ds.root_enabled = True
+    recorded = []
+    original = actions_sql.check_permissions_for_actions
+
+    async def spy(**kwargs):
+        recorded.append(kwargs["actions"])
+        return await original(**kwargs)
+
+    monkeypatch.setattr(actions_sql, "check_permissions_for_actions", spy)
+    try:
+        yield ds, recorded
+    finally:
+        ds.pm.unregister(name="action_hooks")
+
+
+@pytest.mark.asyncio
+async def test_table_page_precomputes_action_permissions(spying_ds):
+    ds, recorded = spying_ds
+    cookies = {"ds_actor": ds.client.actor_cookie({"id": "root"})}
+    response = await ds.client.get("/analytics/users", cookies=cookies)
+    assert response.status_code == 200
+    # The plugin's permission checks were served from the precomputed batch
+    assert "Drop this table (test-plugin)" in response.text
+    assert "Create a table (test-plugin)" in response.text
+    # One batch covered the table-level actions for the table resource,
+    # and one covered the database-level actions for the database resource
+    batches = [batch for batch in recorded if len(batch) > 1]
+    assert any("drop-table" in batch for batch in batches)
+    assert any("create-table" in batch for batch in batches)
+    # The precompute is scoped to actions relevant to each resource:
+    # no global or query-level actions in any batch, and no mixing of
+    # table-level and database-level actions
+    for batch in batches:
+        assert "view-instance" not in batch
+        assert "view-query" not in batch
+        assert not ("drop-table" in batch and "create-table" in batch)
+    # The hook's own allowed() calls hit the cache - no single-action
+    # fallback queries for the actions it checked
+    assert ["drop-table"] not in recorded
+    assert ["create-table"] not in recorded
+
+
+@pytest.mark.asyncio
+async def test_database_page_precomputes_action_permissions(spying_ds):
+    ds, recorded = spying_ds
+    cookies = {"ds_actor": ds.client.actor_cookie({"id": "root"})}
+    response = await ds.client.get("/analytics", cookies=cookies)
+    assert response.status_code == 200
+    assert "Create a table (test-plugin)" in response.text
+    batches = [batch for batch in recorded if len(batch) > 1]
+    assert any("create-table" in batch for batch in batches)
+    # Scoped to database-level actions only
+    for batch in batches:
+        assert "view-instance" not in batch
+        assert "drop-table" not in batch
+    assert ["create-table"] not in recorded
+
+
+@pytest.mark.asyncio
+async def test_cache_does_not_leak_across_requests(counting_ds):
+    ds, plugin = counting_ds
+    cookies = {"ds_actor": ds.client.actor_cookie({"id": "alice"})}
+    response = await ds.client.get("/analytics/users.json", cookies=cookies)
+    assert response.status_code == 200
+    first_request_gathers = plugin.count(actor_id="alice", action="view-table")
+    assert first_request_gathers > 0
+    response = await ds.client.get("/analytics/users.json", cookies=cookies)
+    assert response.status_code == 200
+    # Second request must re-gather (fresh cache), not reuse the first one
+    assert (
+        plugin.count(actor_id="alice", action="view-table") == first_request_gathers * 2
+    )