Feat/permission update

fastapi_admin_kit/admin/admin_template.py

@@ -1,10 +1,13 @@
 """Admin template management and context building."""
 
+import logging
 from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
     pass
 
+logger = logging.getLogger(__name__)
+
 
 class AdminTemplate:
     """Manages Jinja2 template environment and sidebar context."""
@@ -46,11 +49,11 @@ def slugify(s: str) -> str:
         jinja_env.env.filters["slugify"] = slugify
         app.state.admin_jinja_env = jinja_env
 
-    def sidebar_template_kwargs(self, request: Any) -> dict[str, Any]:
+    async def sidebar_template_kwargs(self, request: Any) -> dict[str, Any]:
         """Thin wrapper — returns sidebar kwargs for TemplateResponse contexts."""
-        return self.build_sidebar_context(request)
+        return await self.build_sidebar_context(request)
 
-    def build_sidebar_context(
+    async def build_sidebar_context(
         self,
         request: Any,
         user: Any = None,
@@ -84,28 +87,34 @@ def build_sidebar_context(
             permissions_map = {}
 
             if user and not is_superuser:
+                user_id = snapshot.get("id") if snapshot else getattr(user, "id", None)
                 role_ids = (
                     snapshot.get("role_ids", []) if snapshot else getattr(user, "role_ids", [])
                 )
-                if role_ids:
+                if role_ids or user_id is not None:
                     try:
                         from sqlalchemy import select
-                        from sqlalchemy.orm import Session
 
-                        from fastapi_admin_kit.auth.models import Permission, admin_role_permissions
+                        from fastapi_admin_kit.auth.models import (
+                            Permission,
+                            UserPermission,
+                            admin_role_permissions,
+                        )
+                        from fastapi_admin_kit.db import get_db_session
+
+                        session = get_db_session(request)
 
-                        engine = request.app.state.admin_engine
-                        with Session(engine) as s:
-                            result = s.execute(
+                        # Load permissions from all roles, merge with OR logic
+                        if role_ids:
+                            result = await session.execute(
                                 select(Permission)
                                 .join(
                                     admin_role_permissions,
                                     Permission.id == admin_role_permissions.c.permission_id,
                                 )
-                                .filter(admin_role_permissions.c.role_id.in_(role_ids))
+                                .where(admin_role_permissions.c.role_id.in_(role_ids))
                             )
-                            rows = result.scalars().all()
-                            for perm in rows:
+                            for perm in result.scalars():
                                 if perm.table_name in permissions_map:
                                     existing = permissions_map[perm.table_name]
                                     permissions_map[perm.table_name] = PermissionSet(
@@ -121,10 +130,35 @@ def build_sidebar_context(
                                         can_edit=perm.can_edit,
                                         can_delete=perm.can_delete,
                                     )
-                    except Exception as exc:
-                        import logging
 
-                        logging.getLogger(__name__).debug("Permission query failed: %s", exc)
+                        # Load direct user permission overrides, merge on top
+                        if user_id is not None:
+                            result = await session.execute(
+                                select(UserPermission, Permission)
+                                .join(Permission, UserPermission.permission_id == Permission.id)
+                                .where(UserPermission.user_id == user_id)
+                            )
+                            for up, perm in result:
+                                table = perm.table_name
+                                if table in permissions_map:
+                                    existing = permissions_map[table]
+                                    permissions_map[table] = PermissionSet(
+                                        can_view=existing.can_view or perm.can_view,
+                                        can_create=existing.can_create or perm.can_create,
+                                        can_edit=existing.can_edit or perm.can_edit,
+                                        can_delete=existing.can_delete or perm.can_delete,
+                                    )
+                                else:
+                                    permissions_map[table] = PermissionSet(
+                                        can_view=perm.can_view,
+                                        can_create=perm.can_create,
+                                        can_edit=perm.can_edit,
+                                        can_delete=perm.can_delete,
+                                    )
+                    except Exception as exc:
+                        logger.warning(
+                            "Permission query failed in sidebar fallback: %s", exc, exc_info=True
+                        )
 
         def _item_visible(item: Any) -> bool:
             return (
@@ -189,7 +223,7 @@ def _filter_items(items: list[Any]) -> list[Any]:
             "settings_visible": settings_visible,
         }
 
-    def apply_sidebar_context(self, request: Any, user: Any, context: dict) -> dict:
+    async def apply_sidebar_context(self, request: Any, user: Any, context: dict) -> dict:
         """Inject nav_groups + permissions_map into a template context dict."""
-        context.update(self.build_sidebar_context(request, user=user))
+        context.update(await self.build_sidebar_context(request, user=user))
         return context

fastapi_admin_kit/admin/core.py

@@ -1018,21 +1018,21 @@ def _build_sidebar(self) -> list:
             admin_path=self.router.admin_path,
         )
 
-    def build_sidebar_context(
+    async def build_sidebar_context(
         self,
         request: Any,
         user: Any = None,
         permissions_map: dict | None = None,
     ) -> dict:
         """Build per-request sidebar context (RBAC filter + permissions map)."""
-        return self.template.build_sidebar_context(
+        return await self.template.build_sidebar_context(
             request, user=user, permissions_map=permissions_map
         )
 
-    def sidebar_template_kwargs(self, request: Any) -> dict[str, Any]:
+    async def sidebar_template_kwargs(self, request: Any) -> dict[str, Any]:
         """Thin wrapper — returns sidebar kwargs for TemplateResponse contexts."""
-        return self.template.sidebar_template_kwargs(request)
+        return await self.template.sidebar_template_kwargs(request)
 
-    def apply_sidebar_context(self, request: Any, user: Any, context: dict) -> dict:
+    async def apply_sidebar_context(self, request: Any, user: Any, context: dict) -> dict:
         """Inject nav_groups + permissions_map into a template context dict."""
-        return self.template.apply_sidebar_context(request, user, context)
+        return await self.template.apply_sidebar_context(request, user, context)

fastapi_admin_kit/auth/dependencies.py

@@ -144,3 +144,31 @@ async def require_superuser(
     if not getattr(user, "is_superuser", False):
         raise HTTPException(status_code=403, detail="Superuser access required.")
     return user
+
+
+# ---------------------------------------------------------------------------
+# Permission checker resolver (single source — deduplicated from renderers/factory)
+# ---------------------------------------------------------------------------
+
+
+async def resolve_permission_checker(request: Request) -> Any:
+    """Resolve a ``PermissionChecker`` for the current request.
+
+    Returns ``None`` if the user is not authenticated or the checker cannot be built.
+    This is the canonical implementation — import from here, not from renderers/factory.
+    """
+    from fastapi_admin_kit.auth.identity import get_current_user_from_cookie
+    from fastapi_admin_kit.auth.permissions import PermissionChecker
+
+    user = await get_current_user_from_cookie(request)
+    if user is None:
+        return None
+
+    from fastapi_admin_kit.db import get_db_session
+
+    async_session = get_db_session(request)
+    if async_session is None:
+        return None
+
+    snapshot = getattr(request.state, "admin_user_snapshot", None)
+    return PermissionChecker(session=async_session, user=user, user_snapshot=snapshot)

fastapi_admin_kit/auth/mixins.py

@@ -85,7 +85,11 @@ def set_hasher(cls, hasher: type) -> None:
         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').
+        """Check if this user has a permission by name.
+
+        Supports two formats:
+        - New format: ``"table_name:action"`` (e.g. ``"products:view"``)
+        - Legacy format: ``"table_name_action"`` (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.
@@ -99,15 +103,17 @@ async def has_perm(self, perm_name: str, session: AsyncSession) -> bool:
             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
+        # Prefer ":" separator (unambiguous), fall back to "_" for legacy
+        if ":" in perm_name:
+            table_name, action = perm_name.rsplit(":", 1)
+        else:
+            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"):
@@ -117,19 +123,15 @@ async def has_perm(self, perm_name: str, session: AsyncSession) -> bool:
         if not role_ids and not self.id:
             return False
 
-        # Check role-based permissions
+        # Check role-based permissions (use pre-computed role_ids, not redundant join)
         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)
+                .where(admin_role_permissions.c.role_id.in_(role_ids))
             )
             for perm in result.scalars():
                 if perm.table_name == table_name and getattr(perm, attr, False):

fastapi_admin_kit/auth/permissions.py

@@ -181,8 +181,9 @@ async def get_allowed_fields(self, table_name: str, mode: str) -> set[str] | Non
     def permission_set(self, table_name: str) -> PermissionSet:
         """Return a :class:`PermissionSet` for convenient template / UI use.
 
-        Note: This is a sync convenience wrapper. For async contexts,
-        use the individual async methods directly.
+        Warning: This reads from the async-populated cache. If ``load_permissions()``
+        was not called first, all values will be ``False`` (except for superusers).
+        Always call ``await load_permissions(table_name)`` before using this method.
         """
         if self._is_superuser:
             return PermissionSet(
@@ -191,6 +192,16 @@ def permission_set(self, table_name: str) -> PermissionSet:
                 can_edit=True,
                 can_delete=True,
             )
+        # Check if cache has been populated for this table
+        cache_populated = any(key[0] == table_name for key in self._cache)
+        if not cache_populated:
+            import logging
+
+            logging.getLogger(__name__).warning(
+                "permission_set('%s') called before load_permissions() — "
+                "returning default (all-False). Always call await load_permissions() first.",
+                table_name,
+            )
         return PermissionSet(
             can_view=self._cache.get((table_name, "view"), False),
             can_create=self._cache.get((table_name, "create"), False),

fastapi_admin_kit/cli/permissions.py

@@ -215,13 +215,13 @@ async def _create_admin_permissions(args: argparse.Namespace) -> None:
 
 
 async def _delete_permissions(args: argparse.Namespace) -> None:
-    """Delete all permissions and clear role-permission associations."""
+    """Delete all permissions and clear role-permission and user-permission associations."""
     from sqlalchemy import delete
     from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
     from sqlalchemy.orm import sessionmaker
     from sqlalchemy.pool import NullPool
 
-    from fastapi_admin_kit.auth.models import Permission, admin_role_permissions
+    from fastapi_admin_kit.auth.models import Permission, UserPermission, admin_role_permissions
     from fastapi_admin_kit.models.base import Base
 
     from .user import _resolve_database_url
@@ -238,15 +238,16 @@ async def _delete_permissions(args: argparse.Namespace) -> None:
     async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
 
     async with async_session() as session:
-        # First clear role-permission associations
+        # Clear all associations first
         await session.execute(delete(admin_role_permissions))
+        await session.execute(delete(UserPermission))
         # Then delete all permissions
         await session.execute(delete(Permission))
         await session.commit()
 
     await engine.dispose()
 
-    print("Deleted all permissions (and role-permission associations).")
+    print("Deleted all permissions (and role-permission/user-permission associations).")
 
 
 def register_permission_commands(subparsers) -> None:

fastapi_admin_kit/views/audit.py

@@ -8,22 +8,14 @@
 from sqlalchemy import desc, func, select
 
 from fastapi_admin_kit.audit.models import AuditLog
-from fastapi_admin_kit.auth.dependencies import get_current_admin_user
+from fastapi_admin_kit.auth.dependencies import require_superuser
 from fastapi_admin_kit.auth.protocol import AdminUserProtocol
 from fastapi_admin_kit.db import get_db_session
 from fastapi_admin_kit.views.sidebar import inject_sidebar_context
 
 router = APIRouter()
 
 
-async def _require_superuser(
-    user: AdminUserProtocol = Depends(get_current_admin_user),
-) -> AdminUserProtocol:
-    if not getattr(user, "is_superuser", False):
-        raise HTTPException(status_code=403, detail="Superuser access required.")
-    return user
-
-
 @router.get("/audit-log")
 async def audit_list_view(
     request: Request,
@@ -34,7 +26,7 @@ async def audit_list_view(
     to_date: date | None = Query(None),
     object_id: str | None = Query(None),
     page: int = Query(1, ge=1),
-    _: AdminUserProtocol = Depends(_require_superuser),
+    _: AdminUserProtocol = Depends(require_superuser),
 ):
     """List audit log entries with filters."""
     templates = request.app.state.admin_jinja_env
@@ -116,7 +108,7 @@ async def audit_list_view(
 async def audit_detail_view(
     request: Request,
     entry_id: int,
-    _: AdminUserProtocol = Depends(_require_superuser),
+    _: AdminUserProtocol = Depends(require_superuser),
 ):
     """Show detailed audit entry with diff snapshot."""
     templates = request.app.state.admin_jinja_env

fastapi_admin_kit/views/factory.py

@@ -9,6 +9,9 @@
 from starlette.datastructures import UploadFile
 
 from fastapi_admin_kit.admin.builtin_models import flush_pending_perm_ops
+from fastapi_admin_kit.auth.dependencies import (
+    resolve_permission_checker as _resolve_permission_checker,
+)
 from fastapi_admin_kit.db import get_db_session
 from fastapi_admin_kit.flash import add_flash
 from fastapi_admin_kit.registry import RegisteredModel
@@ -169,30 +172,6 @@ def _get_storage(request: Request):
     return getattr(request.app.state, "admin_storage", None)
 
 
-async def _resolve_permission_checker(request: Request) -> Any:
-    """Resolve a PermissionChecker for the current request.
-
-    Returns None if the user is not authenticated or the checker cannot be built.
-    Delegates current-user resolution to :mod:`fastapi_admin_kit.auth.identity`, the
-    single request-authentication seam — so the user is loaded per request on
-    ``request.state.admin_user`` and never cached on ``app.state`` (which
-    previously leaked identity across concurrent requests).
-    """
-    from fastapi_admin_kit.auth.identity import get_current_user_from_cookie
-    from fastapi_admin_kit.auth.permissions import PermissionChecker
-
-    user = await get_current_user_from_cookie(request)
-    if user is None:
-        return None
-
-    async_session = get_db_session(request)
-    if async_session is None:
-        return None
-
-    snapshot = getattr(request.state, "admin_user_snapshot", None)
-    return PermissionChecker(session=async_session, user=user, user_snapshot=snapshot)
-
-
 async def _handle_file_field(
     request: Request,
     widget: Any,