Simplify types in datasette/permissions.py

This commit is contained in:
Simon Willison 2025-10-23 21:59:46 -07:00
commit 58ac5ccd6e
2 changed files with 21 additions and 28 deletions

View file

@ -2,7 +2,7 @@
from __future__ import annotations
import json
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple, Union
from typing import Any, Dict, Iterable, List, Sequence, Tuple
import sqlite3
from datasette.permissions import PermissionSQL
@ -32,12 +32,8 @@ def _namespace_params(i: int, params: Dict[str, Any]) -> Tuple[str, Dict[str, An
return rewrite, namespaced
PluginProvider = Callable[[str], PermissionSQL]
PluginOrFactory = Union[PermissionSQL, PluginProvider]
def build_rules_union(
actor: Optional[dict], plugins: Sequence[PermissionSQL]
actor: dict | None, plugins: Sequence[PermissionSQL]
) -> Tuple[str, Dict[str, Any]]:
"""
Compose plugin SQL into a UNION ALL with namespaced parameters.
@ -80,11 +76,11 @@ def build_rules_union(
async def resolve_permissions_from_catalog(
db,
actor: Optional[dict],
plugins: Sequence[PluginOrFactory],
actor: dict | None,
plugins: Sequence[Any],
action: str,
candidate_sql: str,
candidate_params: Optional[Dict[str, Any]] = None,
candidate_params: Dict[str, Any] | None = None,
*,
implicit_deny: bool = True,
) -> List[Dict[str, Any]]:
@ -96,8 +92,8 @@ async def resolve_permissions_from_catalog(
(Use child=NULL for parent-scoped actions like "execute-sql".)
- *db* exposes: rows = await db.execute(sql, params)
where rows is an iterable of sqlite3.Row
- plugins are either PermissionSQL objects or callables accepting (action: str)
and returning PermissionSQL instances selecting (parent, child, allow, reason)
- plugins: hook results handled by await_me_maybe - can be sync/async,
single PermissionSQL, list, or callable returning PermissionSQL
- actor is the actor dict (or None), made available as :actor (JSON), :actor_id, and :action
Decision policy:
@ -194,9 +190,9 @@ async def resolve_permissions_from_catalog(
async def resolve_permissions_with_candidates(
db,
actor: Optional[dict],
plugins: Sequence[PluginOrFactory],
candidates: List[Tuple[str, Optional[str]]],
actor: dict | None,
plugins: Sequence[Any],
candidates: List[Tuple[str, str | None]],
action: str,
*,
implicit_deny: bool = True,

View file

@ -1,11 +1,8 @@
import pytest
from datasette.app import Datasette
from datasette.permissions import PermissionSQL
from datasette.utils.permissions import (
PluginProvider,
resolve_permissions_from_catalog,
)
from typing import List
from datasette.utils.permissions import resolve_permissions_from_catalog
from typing import Callable, List
@pytest.fixture
@ -25,7 +22,7 @@ NO_RULES_SQL = (
)
def plugin_allow_all_for_user(user: str) -> PluginProvider:
def plugin_allow_all_for_user(user: str) -> Callable[[str], PermissionSQL]:
def provider(action: str) -> PermissionSQL:
return PermissionSQL(
"allow_all",
@ -40,7 +37,7 @@ def plugin_allow_all_for_user(user: str) -> PluginProvider:
return provider
def plugin_deny_specific_table(user: str, parent: str, child: str) -> PluginProvider:
def plugin_deny_specific_table(user: str, parent: str, child: str) -> Callable[[str], PermissionSQL]:
def provider(action: str) -> PermissionSQL:
return PermissionSQL(
"deny_specific_table",
@ -55,7 +52,7 @@ def plugin_deny_specific_table(user: str, parent: str, child: str) -> PluginProv
return provider
def plugin_org_policy_deny_parent(parent: str) -> PluginProvider:
def plugin_org_policy_deny_parent(parent: str) -> Callable[[str], PermissionSQL]:
def provider(action: str) -> PermissionSQL:
return PermissionSQL(
"org_policy_parent_deny",
@ -69,7 +66,7 @@ def plugin_org_policy_deny_parent(parent: str) -> PluginProvider:
return provider
def plugin_allow_parent_for_user(user: str, parent: str) -> PluginProvider:
def plugin_allow_parent_for_user(user: str, parent: str) -> Callable[[str], PermissionSQL]:
def provider(action: str) -> PermissionSQL:
return PermissionSQL(
"allow_parent",
@ -84,7 +81,7 @@ def plugin_allow_parent_for_user(user: str, parent: str) -> PluginProvider:
return provider
def plugin_child_allow_for_user(user: str, parent: str, child: str) -> PluginProvider:
def plugin_child_allow_for_user(user: str, parent: str, child: str) -> Callable[[str], PermissionSQL]:
def provider(action: str) -> PermissionSQL:
return PermissionSQL(
"allow_child",
@ -99,7 +96,7 @@ def plugin_child_allow_for_user(user: str, parent: str, child: str) -> PluginPro
return provider
def plugin_root_deny_for_all() -> PluginProvider:
def plugin_root_deny_for_all() -> Callable[[str], PermissionSQL]:
def provider(action: str) -> PermissionSQL:
return PermissionSQL(
"root_deny",
@ -114,7 +111,7 @@ def plugin_root_deny_for_all() -> PluginProvider:
def plugin_conflicting_same_child_rules(
user: str, parent: str, child: str
) -> List[PluginProvider]:
) -> List[Callable[[str], PermissionSQL]]:
def allow_provider(action: str) -> PermissionSQL:
return PermissionSQL(
"conflict_child_allow",
@ -140,7 +137,7 @@ def plugin_conflicting_same_child_rules(
return [allow_provider, deny_provider]
def plugin_allow_all_for_action(user: str, allowed_action: str) -> PluginProvider:
def plugin_allow_all_for_action(user: str, allowed_action: str) -> Callable[[str], PermissionSQL]:
def provider(action: str) -> PermissionSQL:
if action != allowed_action:
return PermissionSQL(
@ -475,7 +472,7 @@ async def test_actor_actor_id_action_parameters_available(db):
"""Test that :actor (JSON), :actor_id, and :action are all available in SQL"""
await seed_catalog(db)
def plugin_using_all_parameters() -> PluginProvider:
def plugin_using_all_parameters() -> Callable[[str], PermissionSQL]:
def provider(action: str) -> PermissionSQL:
return PermissionSQL(
"test_all_params",