feat: implement cursor paginantion to role & permission

Author: fiqrikm18

src/core/authorization/domain/service.py

@@ -1,6 +1,8 @@
 from abc import ABC, abstractmethod
+from datetime import datetime
 from uuid import UUID
 
+from src.core.utils.cursor import CursorDirection
 from src.modules.authorization.domain.entities.permission import Permission
 from src.modules.authorization.domain.entities.role import Role
 
@@ -38,6 +40,16 @@ async def get_role(self, role_id: UUID) -> Role | None:
     async def list_roles(self) -> list[Role]:
         pass
 
+    @abstractmethod
+    async def list_roles_cursor(
+        self,
+        cursor_created_at: datetime | None = None,
+        cursor_id: UUID | None = None,
+        limit: int = 10,
+        direction: CursorDirection = CursorDirection.DIRECTION_NEXT,
+    ) -> tuple[list[Role], bool]:
+        pass
+
     @abstractmethod
     async def create_permission(self, permission: Permission) -> Permission:
         pass
@@ -58,6 +70,16 @@ async def get_permission(self, permission_id: UUID) -> Permission | None:
     async def list_permissions(self) -> list[Permission]:
         pass
 
+    @abstractmethod
+    async def list_permissions_cursor(
+        self,
+        cursor_created_at: datetime | None = None,
+        cursor_id: UUID | None = None,
+        limit: int = 10,
+        direction: CursorDirection = CursorDirection.DIRECTION_NEXT,
+    ) -> tuple[list[Permission], bool]:
+        pass
+
     @abstractmethod
     async def assign_permission_to_role(
         self,

src/core/authorization/infrastructure/repositories/casbin_policy_repository.py

@@ -1,8 +1,10 @@
+from datetime import datetime
 from uuid import UUID
 
-from sqlalchemy import delete, select
+from sqlalchemy import and_, delete, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from src.core.utils.cursor import CursorDirection
 from src.core.authorization.infrastructure.models.casbin_rule_model import (
     CasbinRuleModel,
 )
@@ -12,10 +14,10 @@
 from src.core.authorization.infrastructure.models.resource_model import (
     AuthorizationResourceModel,
 )
+from src.core.authorization.infrastructure.models.role_model import RoleModel
 from src.core.authorization.infrastructure.models.role_permission_model import (
     RolePermissionModel,
 )
-from src.core.authorization.infrastructure.models.role_model import RoleModel
 from src.core.authorization.infrastructure.models.user_has_role_model import (
     UserHasRoleModel,
 )
@@ -122,10 +124,7 @@ async def create_resource(
 
     async def list_resources(self) -> list[AuthorizationResource]:
         result = await self._db.execute(select(AuthorizationResourceModel))
-        return [
-            self._resource_from_model(model)
-            for model in result.scalars().all()
-        ]
+        return [self._resource_from_model(model) for model in result.scalars().all()]
 
     async def create_role(self, role: Role) -> Role:
         model = RoleModel(
@@ -138,7 +137,9 @@ async def create_role(self, role: Role) -> Role:
         return self._role_from_model(model)
 
     async def get_role(self, role_id: UUID) -> Role | None:
-        result = await self._db.execute(select(RoleModel).where(RoleModel.id == role_id))
+        result = await self._db.execute(
+            select(RoleModel).where(RoleModel.id == role_id)
+        )
         model = result.scalar_one_or_none()
         if model is None:
             return None
@@ -148,8 +149,36 @@ async def list_roles(self) -> list[Role]:
         result = await self._db.execute(select(RoleModel))
         return [self._role_from_model(model) for model in result.scalars().all()]
 
+    async def list_roles_cursor(
+        self,
+        cursor_created_at: datetime | None = None,
+        cursor_id: UUID | None = None,
+        limit: int = 10,
+        direction: CursorDirection = CursorDirection.DIRECTION_NEXT,
+    ) -> tuple[list[Role], bool]:
+        query = select(RoleModel)
+        query = self._apply_cursor_pagination(
+            query,
+            RoleModel,
+            cursor_created_at,
+            cursor_id,
+            direction,
+        ).limit(limit + 1)
+
+        result = await self._db.execute(query)
+        models = list(result.scalars().all())
+        has_more = len(models) > limit
+        models = models[:limit]
+
+        if direction == CursorDirection.DIRECTION_PREV:
+            models = list(reversed(models))
+
+        return [self._role_from_model(model) for model in models], has_more
+
     async def update_role(self, role: Role) -> Role | None:
-        result = await self._db.execute(select(RoleModel).where(RoleModel.id == role.id))
+        result = await self._db.execute(
+            select(RoleModel).where(RoleModel.id == role.id)
+        )
         model = result.scalar_one_or_none()
         if model is None:
             return None
@@ -209,10 +238,33 @@ async def get_permission(self, permission_id: UUID) -> Permission | None:
 
     async def list_permissions(self) -> list[Permission]:
         result = await self._db.execute(select(PermissionModel))
-        return [
-            self._permission_from_model(model)
-            for model in result.scalars().all()
-        ]
+        return [self._permission_from_model(model) for model in result.scalars().all()]
+
+    async def list_permissions_cursor(
+        self,
+        cursor_created_at: datetime | None = None,
+        cursor_id: UUID | None = None,
+        limit: int = 10,
+        direction: CursorDirection = CursorDirection.DIRECTION_NEXT,
+    ) -> tuple[list[Permission], bool]:
+        query = select(PermissionModel)
+        query = self._apply_cursor_pagination(
+            query,
+            PermissionModel,
+            cursor_created_at,
+            cursor_id,
+            direction,
+        ).limit(limit + 1)
+
+        result = await self._db.execute(query)
+        models = list(result.scalars().all())
+        has_more = len(models) > limit
+        models = models[:limit]
+
+        if direction == CursorDirection.DIRECTION_PREV:
+            models = list(reversed(models))
+
+        return [self._permission_from_model(model) for model in models], has_more
 
     async def update_permission(self, permission: Permission) -> Permission | None:
         result = await self._db.execute(
@@ -287,9 +339,13 @@ async def list_role_permissions(self) -> list[tuple[str, str]]:
         result = await self._db.execute(
             select(RoleModel.name, PermissionModel.key)
             .join(RolePermissionModel, RolePermissionModel.role_id == RoleModel.id)
-            .join(PermissionModel, PermissionModel.id == RolePermissionModel.permission_id)
+            .join(
+                PermissionModel, PermissionModel.id == RolePermissionModel.permission_id
+            )
         )
-        return [(role_name, permission_key) for role_name, permission_key in result.all()]
+        return [
+            (role_name, permission_key) for role_name, permission_key in result.all()
+        ]
 
     async def remove_permission_from_role(
         self,
@@ -319,7 +375,9 @@ def _to_policy_tuple(self, rule: CasbinRuleModel) -> tuple[str, ...]:
         populated = [value for value in values if value is not None]
         return (rule.ptype, *populated)
 
-    async def _get_or_create_resource(self, resource_key: str) -> AuthorizationResourceModel:
+    async def _get_or_create_resource(
+        self, resource_key: str
+    ) -> AuthorizationResourceModel:
         result = await self._db.execute(
             select(AuthorizationResourceModel).where(
                 AuthorizationResourceModel.key == resource_key,
@@ -384,6 +442,8 @@ def _role_from_model(self, model: RoleModel) -> Role:
             id=model.id,
             name=model.name,
             description=model.description,
+            created_at=model.created_at.isoformat(),
+            updated_at=model.updated_at.isoformat(),
         )
 
     def _resource_from_model(
@@ -404,4 +464,40 @@ def _permission_from_model(self, model: PermissionModel) -> Permission:
             resource=model.resource,
             action=model.action,
             description=model.description,
+            created_at=model.created_at.isoformat(),
+            updated_at=model.updated_at.isoformat(),
         )
+
+    def _apply_cursor_pagination(
+        self,
+        query,
+        model,
+        cursor_created_at: datetime | None,
+        cursor_id: UUID | None,
+        direction: CursorDirection,
+    ):
+        if cursor_created_at and cursor_id:
+            if direction == CursorDirection.DIRECTION_NEXT:
+                query = query.where(
+                    or_(
+                        model.created_at < cursor_created_at,
+                        and_(
+                            model.created_at == cursor_created_at,
+                            model.id < cursor_id,
+                        ),
+                    )
+                )
+                return query.order_by(model.created_at.desc(), model.id.desc())
+
+            query = query.where(
+                or_(
+                    model.created_at > cursor_created_at,
+                    and_(
+                        model.created_at == cursor_created_at,
+                        model.id > cursor_id,
+                    ),
+                )
+            )
+            return query.order_by(model.created_at.asc(), model.id.asc())
+
+        return query.order_by(model.created_at.desc(), model.id.desc())

src/core/authorization/infrastructure/services/casbin_authorization_service.py

@@ -1,10 +1,12 @@
+from datetime import datetime
 from uuid import UUID
 
 from src.core.authorization.domain.service import AuthorizationService
 from src.core.authorization.infrastructure.repositories.casbin_policy_repository import (
     SQLAlchemyCasbinPolicyRepository,
 )
 from src.core.authorization.permissions import permission_key
+from src.core.utils.cursor import CursorDirection
 from src.modules.authorization.domain.entities.permission import Permission
 from src.modules.authorization.domain.entities.role import Role
 
@@ -55,6 +57,20 @@ async def get_role(self, role_id: UUID) -> Role | None:
     async def list_roles(self) -> list[Role]:
         return await self._policy_repository.list_roles()
 
+    async def list_roles_cursor(
+        self,
+        cursor_created_at: datetime | None = None,
+        cursor_id: UUID | None = None,
+        limit: int = 10,
+        direction: CursorDirection = CursorDirection.DIRECTION_NEXT,
+    ) -> tuple[list[Role], bool]:
+        return await self._policy_repository.list_roles_cursor(
+            cursor_created_at=cursor_created_at,
+            cursor_id=cursor_id,
+            limit=limit,
+            direction=direction,
+        )
+
     async def create_permission(self, permission: Permission) -> Permission:
         return await self._policy_repository.create_permission(permission)
 
@@ -70,6 +86,20 @@ async def get_permission(self, permission_id: UUID) -> Permission | None:
     async def list_permissions(self) -> list[Permission]:
         return await self._policy_repository.list_permissions()
 
+    async def list_permissions_cursor(
+        self,
+        cursor_created_at: datetime | None = None,
+        cursor_id: UUID | None = None,
+        limit: int = 10,
+        direction: CursorDirection = CursorDirection.DIRECTION_NEXT,
+    ) -> tuple[list[Permission], bool]:
+        return await self._policy_repository.list_permissions_cursor(
+            cursor_created_at=cursor_created_at,
+            cursor_id=cursor_id,
+            limit=limit,
+            direction=direction,
+        )
+
     async def assign_permission_to_role(
         self,
         role_id: UUID,

src/modules/authorization/domain/entities/permission.py

@@ -9,10 +9,16 @@ class Permission:
     resource: str
     action: str
     description: str | None = None
+    created_at: str | None = None
+    updated_at: str | None = None
 
     @classmethod
     def create(
-        cls, key: str, resource: str, action: str, description: str | None
+        cls,
+        key: str,
+        resource: str,
+        action: str,
+        description: str | None,
     ) -> "Permission":
         return cls(
             id=uuid4(),

src/modules/authorization/domain/entities/role.py

@@ -6,7 +6,9 @@
 class Role:
     id: UUID
     name: str
-    description: str | None
+    description: str | None = None
+    created_at: str | None = None
+    updated_at: str | None = None
 
     @classmethod
     def create(cls, name: str, description: str | None = None) -> "Role":