Add permission_for() to create permisisions programatically (#697)

Author: Dreamsorcerer

aiohttp_admin/backends/sqlalchemy.py

@@ -1,14 +1,16 @@
 import asyncio
+import json
 import logging
 import operator
 import sys
+from collections.abc import Sequence
 from types import MappingProxyType
-from typing import Any, Callable, Coroutine, Iterator, Type, TypeVar, Union
+from typing import Any, Callable, Coroutine, Iterator, Literal, Optional, TypeVar, Union
 
 import sqlalchemy as sa
 from aiohttp import web
 from sqlalchemy.ext.asyncio import AsyncEngine
-from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy.orm import DeclarativeBase, QueryableAttribute
 from sqlalchemy.sql.roles import ExpressionElementRole
 
 from .abc import (
@@ -22,6 +24,9 @@
 
 _P = ParamSpec("_P")
 _T = TypeVar("_T")
+_FValues = Union[bool, int, str]
+_Filters = dict[Union[sa.Column[object], QueryableAttribute[Any]],
+                Union[_FValues, Sequence[_FValues]]]
 
 logger = logging.getLogger(__name__)
 
@@ -53,6 +58,58 @@ async def inner(*args: _P.args, **kwargs: _P.kwargs) -> _T:
     return inner
 
 
+def permission_for(sa_obj: Union[sa.Table, type[DeclarativeBase],
+                                 sa.Column[object], QueryableAttribute[Any]],
+                   perm_type: Literal["view", "edit", "add", "delete", "*"] = "*",
+                   *, filters: Optional[_Filters] = None, negated: bool = False) -> str:
+    """Returns a permission string for the given sa_obj.
+
+    Args:
+        sa_obj: A SQLAlchemy object to grant permission to (table/model/column/attribute).
+        perm_type: The type of permission to grant acces to.
+        filters: Filters to restrict the permisson to (can't be used with negated).
+                 e.g. {User.type: "admin", User.active: True} only permits access if
+                      `User.type == "admin" and User.active`.
+                      {Post.type: ("news", "sports")} only permits access if
+                      `Post.type in ("news", "sports")`.
+        negated: True if result should restrict access from sa_obj.
+    """
+    if filters and negated:
+        raise ValueError("Can't use filters on negated permissions.")
+    if perm_type not in {"view", "edit", "add", "delete", "*"}:
+        raise ValueError(f"Invalid perm_type: '{perm_type}'")
+
+    field = None
+    if isinstance(sa_obj, sa.Table):
+        table = sa_obj
+    elif isinstance(sa_obj, (sa.Column, QueryableAttribute)):
+        table = sa_obj.table
+        field = sa_obj.name
+    else:
+        if not isinstance(sa_obj.__table__, sa.Table):
+            raise ValueError("Non-table mappings are not supported.")
+        table = sa_obj.__table__
+    p = "{}admin.{}".format("~" if negated else "", table.name)
+
+    if field:
+        p = f"{p}.{field}"
+
+    p = f"{p}.{perm_type}"
+
+    if filters:
+        for col, value in filters.items():
+            if col.table is not table:
+                raise ValueError("Filter key not an attribute/column of sa_obj.")
+            # Sequences should be treated as multiple filter values for that key.
+            if not isinstance(value, Sequence) or isinstance(value, str):
+                value = (value,)
+            for v in value:
+                v = json.dumps(v)
+                p += f"|{col.name}={v}"
+
+    return p
+
+
 def create_filters(columns: sa.ColumnCollection[str, sa.Column[object]],
                    filters: dict[str, object]) -> Iterator[ExpressionElementRole[Any]]:
     return (columns[k].in_(v) if isinstance(v, list)
@@ -61,7 +118,7 @@ def create_filters(columns: sa.ColumnCollection[str, sa.Column[object]],
 
 
 class SAResource(AbstractAdminResource):
-    def __init__(self, db: AsyncEngine, model_or_table: Union[sa.Table, Type[DeclarativeBase]]):
+    def __init__(self, db: AsyncEngine, model_or_table: Union[sa.Table, type[DeclarativeBase]]):
         if isinstance(model_or_table, sa.Table):
             table = model_or_table
         else:

examples/permissions.py

@@ -16,7 +16,7 @@
 import aiohttp_admin
 from _models import Base, Simple, SimpleParent
 from aiohttp_admin import Permissions, UserDetails
-from aiohttp_admin.backends.sqlalchemy import SAResource
+from aiohttp_admin.backends.sqlalchemy import SAResource, permission_for as p
 
 
 class User(Base):
@@ -48,14 +48,14 @@ async def create_app() -> web.Application:
         await conn.run_sync(Base.metadata.create_all)
     async with session.begin() as sess:
         sess.add(Simple(num=5, value="first"))
-        p = Simple(num=82, optional_num=12, value="with child")
-        sess.add(p)
+        p_simple = Simple(num=82, optional_num=12, value="with child")
+        sess.add(p_simple)
         sess.add(Simple(num=5, value="second"))
         sess.add(Simple(num=5, value="3"))
         sess.add(Simple(num=5, optional_num=42, value="4"))
         sess.add(Simple(num=5, value="5"))
     async with session.begin() as sess:
-        sess.add(SimpleParent(id=p.id, date=datetime(2023, 2, 13, 19, 4)))
+        sess.add(SimpleParent(id=p_simple.id, date=datetime(2023, 2, 13, 19, 4)))
 
     app = web.Application()
     app["db"] = session
@@ -86,17 +86,19 @@ async def create_app() -> web.Application:
         sess.add(User(username="delete", permissions=json.dumps(
             (Permissions.view, Permissions.delete))))
         users = {
-            "simple": ("admin.simple.*",),
-            "mixed": ("admin.simple.view", "admin.simple.edit", "admin.parent.view"),
-            "negated": ("admin.*", "~admin.parent.*", "~admin.simple.edit"),
-            "field": ("admin.*", "~admin.simple.optional_num.*"),
-            "field_edit": ("admin.*", "~admin.simple.optional_num.edit"),
-            "filter": ("admin.*", "admin.simple.*|num=5"),
-            "filter_edit": ("admin.*", "admin.simple.edit|num=5"),
-            "filter_add": ("admin.*", "admin.simple.add|num=5"),
-            "filter_delete": ("admin.*", "admin.simple.delete|num=5"),
-            "filter_field": ("admin.*", "admin.simple.optional_num.*|num=5"),
-            "filter_field_edit": ("admin.*", "admin.simple.optional_num.edit|num=5")
+            "simple": (p(Simple),),
+            "mixed": (p(Simple, "view"), p(Simple, "edit"), p(SimpleParent, "view")),
+            "negated": (Permissions.all, p(SimpleParent, negated=True),
+                        p(Simple, "edit", negated=True)),
+            "field": (Permissions.all, p(Simple.optional_num, negated=True)),
+            "field_edit": (Permissions.all, p(Simple.optional_num, "edit", negated=True)),
+            "filter": (Permissions.all, p(Simple, filters={Simple.num: 5})),
+            "filter_edit": (Permissions.all, p(Simple, "edit", filters={Simple.num: 5})),
+            "filter_add": (Permissions.all, p(Simple, "add", filters={Simple.num: 5})),
+            "filter_delete": (Permissions.all, p(Simple, "delete", filters={Simple.num: 5})),
+            "filter_field": (Permissions.all, p(Simple.optional_num, filters={Simple.num: 5})),
+            "filter_field_edit": (Permissions.all, p(Simple.optional_num, "edit",
+                                                     filters={Simple.num: 5}))
         }
         for name, permissions in users.items():
             if any(admin["permission_re"].fullmatch(p) is None for p in permissions):

tests/test_backends_sqlalchemy.py

@@ -11,7 +11,7 @@
 
 import aiohttp_admin
 from _auth import check_credentials
-from aiohttp_admin.backends.sqlalchemy import SAResource
+from aiohttp_admin.backends.sqlalchemy import SAResource, permission_for
 
 _Login = Callable[[TestClient], Awaitable[dict[str, str]]]
 
@@ -285,3 +285,37 @@ class TestModel(base):  # type: ignore[misc,valid-type]
         assert resp.status == 200
         assert await resp.json() == {"data": {"id": 2, "date": "2024-05-09",
                                               "time": "2020-11-12 03:04:05"}}
+
+
+def test_permission_for(base: type[DeclarativeBase]) -> None:
+    class M(base):  # type: ignore[misc,valid-type]
+        __tablename__ = "test"
+        id: Mapped[int] = mapped_column(primary_key=True)
+        cat: Mapped[int]
+        val: Mapped[str]
+
+    t = M.__table__
+
+    assert permission_for(M) == "admin.test.*"
+    assert permission_for(M, "view") == "admin.test.view"
+    assert permission_for(M, "add", negated=True) == "~admin.test.add"
+    assert permission_for(M.cat, "edit") == "admin.test.cat.edit"
+    assert permission_for(t.c["val"], "*", negated=True) == "~admin.test.val.*"
+    assert permission_for(M, filters={M.cat: 5, M.val: "Foo"}) == 'admin.test.*|cat=5|val="Foo"'
+    assert permission_for(
+        t, "delete", filters={t.c["val"]: "bar"}) == 'admin.test.delete|val="bar"'
+    assert permission_for(M.val, filters={M.id: (3, 4)}) == "admin.test.val.*|id=3|id=4"
+    assert permission_for(
+        M.cat, "edit", filters={M.cat: [1, 5]}) == "admin.test.cat.edit|cat=1|cat=5"
+
+    with pytest.raises(ValueError, match="Can't use filters on negated"):
+        permission_for(M, filters={M.id: 1}, negated=True)
+    with pytest.raises(ValueError, match="foo"):
+        permission_for(M, "foo")  # type: ignore[arg-type]
+
+    class Wrong(base):  # type: ignore[misc,valid-type]
+        __tablename__ = "wrong"
+        id: Mapped[int] = mapped_column(primary_key=True)
+
+    with pytest.raises(ValueError, match="not an attribute"):
+        permission_for(M, filters={Wrong.id: 1})