Request-scoped permission check cache

Adds a per-request cache for permission check results, plus wiring that
resolves action permissions in bulk before plugin hooks need them:

- New _permission_check_cache contextvar, set to a fresh dict for each
  request by DatasetteRouter and reset when the request ends. Keys
  include the full serialized actor, so actors differing in any field
  (e.g. token restrictions) never share entries. SkipPermissions mode
  bypasses the cache entirely.
- datasette.allowed_many() now consults the cache and stores its
  results there, so repeated datasette.allowed() checks within one
  request resolve without further SQL.
- Table pages resolve all registered table-level actions against the
  current table and all database-level actions against its database
  (database pages likewise) in batched queries before invoking the
  table_actions/database_actions plugin hooks - allowed() calls made
  inside those hooks are then served from the cache with no plugin
  changes required. Actions with no permission rules from any plugin
  are resolved to False without touching the database.

Benchmarks (benchmarks/) with a simulated 12-plugin ecosystem making
18 checks per table page show 34 -> 13 internal-DB queries per page;
with 2ms-per-query internal DB latency (modelling Datasette Cloud)
table page time drops from 77.9ms to 27.6ms - the caching layer
accounts for ~91% of that improvement over allowed_many() alone.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

datasette/app.py

@@ -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,
@@ -1834,7 +1843,9 @@ class Datasette:
 
         Resolves every action (plus any also_requires dependencies) with a
         single internal database query, instead of one or two queries per
-        action.
+        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
@@ -1846,6 +1857,10 @@ class Datasette:
             # {"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
@@ -1869,14 +1884,30 @@ class Datasette:
         for name in requested:
             add_action(name)
 
-        raw = await check_permissions_for_actions(
-            datasette=self,
-            actor=actor,
-            actions=expanded,
-            parent=parent,
-            child=child,
-        )
+        # 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,
+                actions=to_check,
+                parent=parent,
+                child=child,
+            )
 
         def resolve(name):
             # final verdict = own rules AND verdict of also_requires chain
@@ -1892,8 +1923,13 @@ class Datasette:
         for name in expanded:
             resolve(name)
 
-        # Log every check for the debug page, dependencies before the
-        # actions that required them
+        # 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(
@@ -2663,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/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,8 @@ 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)
@@ -543,6 +545,8 @@ Example usage:
     )
     # {"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:

docs/plugin_hooks.rst

@@ -1460,9 +1460,9 @@ plugin's source name (e.g., ``myplugin_user_id``). The system reserves these par
 
 This hook may be called for many actions in rapid succession - for example
 :ref:`datasette.allowed_many() <datasette_allowed_many>` gathers rules for every action in its batch
-concurrently. Hook implementations must not assume that checks for different actions arrive one
-page-render apart, and expensive work (such as network calls) should be cached independently of the
-``action`` argument where possible.
+concurrently before table and database pages render their action menus. Hook implementations must not
+assume that checks for different actions arrive one page-render apart, and expensive work (such as
+network calls) should be cached independently of the ``action`` argument where possible.
 
 You can also use return ``PermissionSQL.allow(reason="reason goes here")`` or ``PermissionSQL.deny(reason="reason goes here")`` as shortcuts for simple root-level allow or deny rules. These will create SQL snippets that look like this:
 

tests/test_allowed_many.py

@@ -1,18 +1,52 @@
 """
-Tests for the datasette.allowed_many() batch permission API, which
-resolves multiple actions against one resource in a single internal
-database query. datasette.allowed() is implemented on top of it, so
-both entry points share one resolution code path.
+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 PermissionSQL, SkipPermissions
+from datasette.permissions import (
+    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()
@@ -24,6 +58,154 @@ async def ds():
     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."""
 
@@ -233,7 +415,7 @@ class ParamCollisionPlugin:
 
 @pytest.mark.asyncio
 async def test_allowed_many_namespaces_params_across_actions(ds):
-    """Many actions whose rules use identical param names must not collide."""
+    """40+ actions whose rules use identical param names must not collide."""
     plugin = ParamCollisionPlugin()
     ds.pm.register(plugin, name="collision")
     try:
@@ -330,6 +512,24 @@ async def test_allowed_many_global_actions_without_resource(ds):
     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():
@@ -339,3 +539,131 @@ async def test_allowed_many_skip_permission_checks(ds):
             actor=None,
         )
     assert results == {"view-table": True, "drop-table": True}
+
+
+# ----------------------------------------------------------------------
+# 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
+    )