feat: Refactor permission management to use permission IDs instead of table-based permissions, enhancing user permission handling and UI integration

Author: borhanst

fastapi_admin_kit/admin/builtin_models.py

@@ -7,24 +7,8 @@
 
 
 async def flush_pending_perm_ops(request):
-    """Execute any pending direct-permission writes on the request's session.
-
-    Called after ``after_create`` / ``after_update`` so the ops run on the
-    same session (and thus the same SQLite connection) as the main request.
-    """
-    from sqlalchemy import text
-
-    from fastapi_admin_kit.db import get_db_session
-
-    ops = getattr(request.state, "_admin_perm_pending_ops", None)
-    if not ops:
-        return
-    request.state._admin_perm_pending_ops = []
-    session = get_db_session(request)
-    if session is None:
-        return
-    for sql_str, params in ops:
-        await session.execute(text(sql_str), params)
+    """No-op — direct permissions are now saved immediately."""
+    pass
 
 
 def _get_table_names() -> list[str]:
@@ -115,66 +99,72 @@ def after_update(self, obj, request=None):
             self._save_direct_permissions_after_commit(obj, perm_data, request)
 
     def _save_direct_permissions_after_commit(self, obj, perm_data, request):
-        delete_sql = "DELETE FROM admin_user_permissions WHERE user_id = :uid"
-        insert_sql = (
-            "INSERT INTO admin_user_permissions"
-            " (user_id, table_name, can_view, can_create, can_edit, can_delete)"
-            " VALUES (:uid, :tn, :cv, :cc, :ce, :cd)"
-        )
-
-        ops = []
-        ops.append((delete_sql, {"uid": obj.id}))
-
-        for table_name, perms in perm_data.items():
-            if not any(perms.get(a) for a in ["view", "create", "edit", "delete"]):
-                continue
-            ops.append(
-                (
-                    insert_sql,
-                    {
-                        "uid": obj.id,
-                        "tn": table_name,
-                        "cv": 1 if perms.get("view") else 0,
-                        "cc": 1 if perms.get("create") else 0,
-                        "ce": 1 if perms.get("edit") else 0,
-                        "cd": 1 if perms.get("delete") else 0,
-                    },
-                )
-            )
+        """Save direct user permissions after the user object is committed.
+
+        perm_data: list of permission IDs, e.g. [1, 3, 5]
+        """
+        import asyncio
+
+        from sqlalchemy import delete
+
+        from fastapi_admin_kit.auth.models import UserPermission
+
+        perm_ids = perm_data if isinstance(perm_data, list) else []
+
+        async def _do_save():
+            from fastapi_admin_kit.db import get_db_session
+
+            session = get_db_session(request)
+            # Delete existing direct permissions for this user
+            await session.execute(delete(UserPermission).where(UserPermission.user_id == obj.id))
 
-        if not hasattr(request.state, "_admin_perm_pending_ops"):
-            request.state._admin_perm_pending_ops = []
-        request.state._admin_perm_pending_ops.extend(ops)
+            # Insert new direct permissions
+            for perm_id in perm_ids:
+                up = UserPermission(user_id=obj.id, permission_id=perm_id)
+                session.add(up)
+
+            await session.commit()
+
+        try:
+            loop = asyncio.get_event_loop()
+            if loop.is_running():
+                asyncio.ensure_future(_do_save())
+            else:
+                loop.run_until_complete(_do_save())
+        except RuntimeError:
+            asyncio.run(_do_save())
 
     async def get_form_context(self, context, obj=None, request=None):
         """Load direct permissions for the user being edited."""
         from sqlalchemy import select
 
-        from fastapi_admin_kit.auth.models import UserPermission
+        from fastapi_admin_kit.auth.models import Permission, UserPermission
         from fastapi_admin_kit.db import get_db_session
 
-        perm_data = {}
+        perm_data = []
         if obj is not None and request is not None:
             try:
                 session = get_db_session(request)
                 result = await session.execute(
-                    select(UserPermission).where(UserPermission.user_id == obj.id)
+                    select(UserPermission, Permission)
+                    .join(Permission, UserPermission.permission_id == Permission.id)
+                    .where(UserPermission.user_id == obj.id)
                 )
-                for p in result.scalars():
-                    perm_data[p.table_name] = {
-                        "_label": p.table_name,
-                        "view": p.can_view,
-                        "create": p.can_create,
-                        "edit": p.can_edit,
-                        "delete": p.can_delete,
-                    }
+                for up, perm in result:
+                    perm_data.append(
+                        {
+                            "id": perm.id,
+                            "name": perm.name,
+                            "table_name": perm.table_name,
+                        }
+                    )
             except Exception as exc:
                 import logging
 
                 logging.getLogger(__name__).debug("Permission load failed: %s", exc)
 
         context["perm_data"] = perm_data
-        context["search_url"] = "/admin/tables/search"
+        context["search_url"] = "/admin/permissions/search"
         return context
 
     def process_form_data(self, data, request=None):

fastapi_admin_kit/api/auth.py

@@ -94,9 +94,11 @@ async def _build_user_permissions(user: Any, db_session: Any) -> dict[str, list[
     user_id = getattr(user, "id", None)
     if user_id is not None:
         result = await db_session.execute(
-            select(UserPermission).where(UserPermission.user_id == user_id)
+            select(UserPermission, Permission)
+            .join(Permission, UserPermission.permission_id == Permission.id)
+            .where(UserPermission.user_id == user_id)
         )
-        for perm in result.scalars():
+        for up, perm in result:
             actions = []
             if perm.can_view:
                 actions.append("view")

fastapi_admin_kit/auth/mixins.py

@@ -7,7 +7,7 @@
 from sqlalchemy import Boolean, Column, String
 
 if TYPE_CHECKING:
-    pass
+    from sqlalchemy.ext.asyncio import AsyncSession
 
 
 class AuthModelMixin:
@@ -42,6 +42,7 @@ class MyUser(AutoModelMixin, Base):
     - ``verify_password(password)`` → bool
     - ``hash_password(password)`` → str (classmethod)
     - ``set_hasher(hasher)`` classmethod
+    - ``has_perm(perm_name, session)`` → bool (check permission by name)
     """
 
     _hasher: ClassVar[type | None] = None
@@ -83,5 +84,68 @@ def set_hasher(cls, hasher: type) -> None:
         """Set the password hasher class for this model."""
         cls._hasher = hasher
 
+    async def has_perm(self, perm_name: str, session: AsyncSession) -> bool:
+        """Check if this user has a permission by name (e.g. 'products_view').
+
+        Returns True if any assigned role grants this permission,
+        or if a direct user permission grants it. Superusers always return True.
+        """
+        if self.is_superuser:
+            return True
+
+        from sqlalchemy import select
+
+        from fastapi_admin_kit.auth.models import (
+            Permission,
+            UserPermission,
+            admin_role_permissions,
+            admin_user_roles,
+        )
+
+        # Parse perm_name -> (table_name, action)
+        # e.g. "products_view" -> ("products", "view")
+        parts = perm_name.rsplit("_", 1)
+        if len(parts) != 2:
+            return False
+        table_name, action = parts
+
+        attr = f"can_{action}"
+        if attr not in ("can_view", "can_create", "can_edit", "can_delete"):
+            return False
+
+        role_ids = self.role_ids
+        if not role_ids and not self.id:
+            return False
+
+        # Check role-based permissions
+        if role_ids:
+            result = await session.execute(
+                select(Permission)
+                .join(
+                    admin_role_permissions,
+                    Permission.id == admin_role_permissions.c.permission_id,
+                )
+                .join(
+                    admin_user_roles,
+                    admin_role_permissions.c.role_id == admin_user_roles.c.role_id,
+                )
+                .where(admin_user_roles.c.user_id == self.id)
+            )
+            for perm in result.scalars():
+                if perm.table_name == table_name and getattr(perm, attr, False):
+                    return True
+
+        # Check direct user permissions
+        result = await session.execute(
+            select(Permission)
+            .join(UserPermission, UserPermission.permission_id == Permission.id)
+            .where(UserPermission.user_id == self.id)
+        )
+        for perm in result.scalars():
+            if perm.table_name == table_name and getattr(perm, attr, False):
+                return True
+
+        return False
+
 
 # Backward-compatible alias

fastapi_admin_kit/auth/models.py

@@ -144,7 +144,7 @@ class UserPermission(Base):
 
     __tablename__ = "admin_user_permissions"
     __table_args__ = (
-        UniqueConstraint("user_id", "table_name", name="uq_admin_user_perm_user_table"),
+        UniqueConstraint("user_id", "permission_id", name="uq_admin_user_perm_user_perm"),
     )
 
     id = Column(Integer, primary_key=True)
@@ -153,19 +153,22 @@ class UserPermission(Base):
         ForeignKey("admin_users.id", ondelete="CASCADE"),
         nullable=False,
     )
-    table_name = Column(String(255), nullable=False)
-    can_view = Column(Boolean, default=False)
-    can_create = Column(Boolean, default=False)
-    can_edit = Column(Boolean, default=False)
-    can_delete = Column(Boolean, default=False)
+    permission_id = Column(
+        Integer,
+        ForeignKey("admin_permissions.id", ondelete="CASCADE"),
+        nullable=False,
+    )
 
     user = relationship("User", back_populates="direct_permissions")
+    permission = relationship("Permission", backref="user_overrides")
 
     def __str__(self) -> str:
-        return f"{self.table_name} (user {self.user_id})"
+        if self.permission:
+            return f"{self.permission.name} (user {self.user_id})"
+        return f"user {self.user_id}"
 
     def __repr__(self) -> str:
-        return f"<UserPermission user={self.user_id} table={self.table_name!r}>"
+        return f"<UserPermission user={self.user_id} perm={self.permission_id}>"
 
 
 class RefreshToken(Base):

fastapi_admin_kit/auth/permissions.py

@@ -96,10 +96,14 @@ async def _load_direct_permissions(self) -> dict[str, PermissionSet | None]:
 
         from sqlalchemy import select
 
+        from fastapi_admin_kit.auth.models import Permission
+
         result = await self.session.execute(
-            select(UserPermission).where(UserPermission.user_id == self._user_id)
+            select(UserPermission, Permission)
+            .join(Permission, UserPermission.permission_id == Permission.id)
+            .where(UserPermission.user_id == self._user_id)
         )
-        for perm in result.scalars():
+        for up, perm in result:
             table = perm.table_name
             if table not in self._direct_cache:
                 self._direct_cache[table] = PermissionSet()

fastapi_admin_kit/auth/views.py

@@ -113,7 +113,6 @@ async def login_post(
     auth_backend: AuthBackend = request.app.state.admin_auth_backend
     login_field = request.app.state.admin_config.get("login_field", "email")
     user = await auth_backend.authenticate(username, password, session, login_field=login_field)
-    print("render page", user)
     if user is not None:
         _login_rate_limiter.reset(client_ip)
         user.last_login = datetime.now(UTC)

fastapi_admin_kit/static/js/admin.js

@@ -205,19 +205,18 @@ document.addEventListener('alpine:init', () => {
   /* ── Permission Widget ────────────────────────────────────────────── */
 
   Alpine.data('permissionWidget', (searchUrl, initialPermData) => ({
-    selectedTables: [],
+    selectedPerms: [],
     searchQuery: '',
     results: [],
     open: false,
     _debounce: null,
-    permData: {},
-    expandedTable: null,
 
     init() {
-      this.permData = initialPermData || {};
-      this.selectedTables = Object.keys(this.permData).map(k => ({
-        id: k, label: this.permData[k]._label || k
-      }));
+      if (initialPermData && Array.isArray(initialPermData)) {
+        this.selectedPerms = initialPermData.map(p => ({
+          id: p.id, name: p.name, table_name: p.table_name
+        }));
+      }
     },
 
     async search() {
@@ -229,58 +228,31 @@ document.addEventListener('alpine:init', () => {
           const resp = await fetch(url);
           if (resp.ok) {
             const all = await resp.json();
-            const selected = new Set(this.selectedTables.map(t => t.id));
+            const selected = new Set(this.selectedPerms.map(p => p.id));
             this.results = all.filter(r => !selected.has(r.id));
           }
         } catch (e) { console.error('Permission search error:', e); }
       }, 250);
     },
 
-    addTable(table) {
-      if (!this.permData[table.id]) {
-        this.permData[table.id] = {
-          _label: table.label,
-          view: false, create: false, edit: false, delete: false
-        };
+    addPerm(perm) {
+      if (!this.selectedPerms.find(p => p.id === perm.id)) {
+        this.selectedPerms.push({ id: perm.id, name: perm.name, table_name: perm.table_name });
       }
-      this.selectedTables.push(table);
       this.searchQuery = '';
       this.results = [];
-      this.expandedTable = table.id;
-    },
-
-    removeTable(index) {
-      const table = this.selectedTables[index];
-      delete this.permData[table.id];
-      this.selectedTables.splice(index, 1);
-      if (this.expandedTable === table.id) this.expandedTable = null;
-    },
-
-    toggleExpand(tableId) {
-      this.expandedTable = this.expandedTable === tableId ? null : tableId;
     },
 
-    toggleAllActions(tableId, on) {
-      this.permData[tableId].view = on;
-      this.permData[tableId].create = on;
-      this.permData[tableId].edit = on;
-      this.permData[tableId].delete = on;
+    removePerm(index) {
+      this.selectedPerms.splice(index, 1);
     },
 
-    toggleAllTables(action, on) {
-      for (const table of this.selectedTables) {
-        if (this.permData[table.id]) {
-          this.permData[table.id][action] = on;
-        }
-      }
+    toggleAll(action, on) {
+      // No-op — permissions are individual, no table grouping
     },
 
     get serializedPermData() {
-      const out = {};
-      for (const [table, data] of Object.entries(this.permData)) {
-        out[table] = { view: data.view, create: data.create, edit: data.edit, delete: data.delete };
-      }
-      return JSON.stringify(out);
+      return JSON.stringify(this.selectedPerms.map(p => p.id));
     }
   }));