feat(BA-5769): add my-role SDK, CLI, and v2 component tests

changes/11179.feature.md

@@ -0,0 +1 @@
+Add my-role REST v2 SDK endpoint and CLI command for users to query their own role assignments

src/ai/backend/client/cli/v2/my/__init__.py

@@ -21,6 +21,11 @@ def keypair() -> None:
     """My keypair commands."""
 
 
+@my.group(cls=LazyGroup, import_name="ai.backend.client.cli.v2.my.role:role")
+def role() -> None:
+    """My role commands."""
+
+
 @my.group(
     cls=LazyGroup,
     import_name="ai.backend.client.cli.v2.my.login_history:login_history",

src/ai/backend/client/cli/v2/my/role.py

@@ -0,0 +1,85 @@
+"""CLI commands for self-service role operations."""
+
+from __future__ import annotations
+
+import asyncio
+
+import click
+
+from ai.backend.client.cli.v2.helpers import (
+    create_v2_registry,
+    load_v2_config,
+    parse_order_options,
+    print_result,
+)
+
+
+@click.group()
+def role() -> None:
+    """My role commands."""
+
+
+@role.command()
+@click.option("--limit", default=None, type=int, help="Maximum number of results to return.")
+@click.option("--offset", default=None, type=int, help="Number of results to skip.")
+@click.option("--first", default=None, type=int, help="Cursor-based: return first N items.")
+@click.option("--after", default=None, type=str, help="Cursor-based: return items after cursor.")
+@click.option("--last", default=None, type=int, help="Cursor-based: return last N items.")
+@click.option("--before", default=None, type=str, help="Cursor-based: return items before cursor.")
+@click.option("--role-id", type=str, default=None, help="Filter by role UUID.")
+@click.option(
+    "--order-by",
+    multiple=True,
+    help="Order by field:direction (e.g., granted_at:desc).",
+)
+def search(
+    limit: int | None,
+    offset: int | None,
+    first: int | None,
+    after: str | None,
+    last: int | None,
+    before: str | None,
+    role_id: str | None,
+    order_by: tuple[str, ...],
+) -> None:
+    """Search my role assignments."""
+    from uuid import UUID
+
+    from ai.backend.common.dto.manager.v2.rbac.request import (
+        RoleAssignmentFilter,
+        RoleAssignmentOrderBy,
+        SearchRoleAssignmentsInput,
+    )
+    from ai.backend.common.dto.manager.v2.rbac.types import RoleAssignmentOrderField
+
+    filter_dto: RoleAssignmentFilter | None = None
+    if role_id is not None:
+        filter_dto = RoleAssignmentFilter(
+            role_id=UUID(role_id),
+        )
+
+    orders = (
+        parse_order_options(order_by, RoleAssignmentOrderField, RoleAssignmentOrderBy)
+        if order_by
+        else None
+    )
+
+    async def _run() -> None:
+        registry = await create_v2_registry(load_v2_config())
+        try:
+            request = SearchRoleAssignmentsInput(
+                filter=filter_dto,
+                order=orders,
+                first=first,
+                after=after,
+                last=last,
+                before=before,
+                limit=limit,
+                offset=offset,
+            )
+            result = await registry.rbac.my_search_assignments(request)
+            print_result(result)
+        finally:
+            await registry.close()
+
+    asyncio.run(_run())

src/ai/backend/client/cli/v2/rbac/assignment.py

@@ -42,9 +42,9 @@ def search(
     """Search role assignments."""
     from ai.backend.common.dto.manager.query import StringFilter
     from ai.backend.common.dto.manager.v2.rbac.request import (
-        AdminSearchRoleAssignmentsGQLInput,
         RoleAssignmentFilter,
         RoleAssignmentOrderBy,
+        SearchRoleAssignmentsInput,
     )
     from ai.backend.common.dto.manager.v2.rbac.types import RoleAssignmentOrderField
 
@@ -70,7 +70,7 @@ async def _run() -> None:
         registry = await create_v2_registry(load_v2_config())
         try:
             result = await registry.rbac.search_assignments(
-                AdminSearchRoleAssignmentsGQLInput(
+                SearchRoleAssignmentsInput(
                     filter=filter_dto,
                     order=orders,
                     limit=limit,

src/ai/backend/client/v2/domains_v2/rbac.py

@@ -8,7 +8,6 @@
 from ai.backend.common.dto.manager.v2.rbac.request import (
     AdminSearchEntitiesGQLInput,
     AdminSearchPermissionsGQLInput,
-    AdminSearchRoleAssignmentsGQLInput,
     AssignRoleInput,
     BulkAssignRoleInput,
     BulkRevokeRoleInput,
@@ -18,14 +17,14 @@
     DeleteRoleInput,
     PurgeRoleInput,
     RevokeRoleInput,
+    SearchRoleAssignmentsInput,
     SearchRolesInput,
     UpdatePermissionInput,
     UpdateRoleInput,
 )
 from ai.backend.common.dto.manager.v2.rbac.response import (
     AdminSearchAssociationsPayload,
     AdminSearchPermissionsPayload,
-    AdminSearchRoleAssignmentsPayload,
     AdminSearchRolesPayload,
     BulkAssignRoleResultPayload,
     BulkRevokeRoleResultPayload,
@@ -36,6 +35,7 @@
     PurgeRolePayload,
     RoleAssignmentNode,
     RoleNode,
+    SearchRoleAssignmentsPayload,
     UpdateRolePayload,
 )
 
@@ -172,14 +172,25 @@ async def revoke_role(self, request: RevokeRoleInput) -> RoleAssignmentNode:
         )
 
     async def search_assignments(
-        self, request: AdminSearchRoleAssignmentsGQLInput
-    ) -> AdminSearchRoleAssignmentsPayload:
+        self, request: SearchRoleAssignmentsInput
+    ) -> SearchRoleAssignmentsPayload:
         """Search role assignments with filters, orders, and pagination."""
         return await self._client.typed_request(
             "POST",
             f"{_PATH}/assignments/search",
             request=request,
-            response_model=AdminSearchRoleAssignmentsPayload,
+            response_model=SearchRoleAssignmentsPayload,
+        )
+
+    async def my_search_assignments(
+        self, request: SearchRoleAssignmentsInput
+    ) -> SearchRoleAssignmentsPayload:
+        """Search role assignments for the current authenticated user."""
+        return await self._client.typed_request(
+            "POST",
+            f"{_PATH}/assignments/my/search",
+            request=request,
+            response_model=SearchRoleAssignmentsPayload,
         )
 
     async def bulk_assign_role(self, request: BulkAssignRoleInput) -> BulkAssignRoleResultPayload:

src/ai/backend/common/dto/manager/v2/rbac/__init__.py

@@ -5,7 +5,6 @@
 from ai.backend.common.dto.manager.v2.rbac.request import (
     AdminSearchEntitiesGQLInput,
     AdminSearchPermissionsGQLInput,
-    AdminSearchRoleAssignmentsGQLInput,
     AssignRoleInput,
     BulkAssignRoleInput,
     BulkRevokeRoleInput,
@@ -24,6 +23,7 @@
     RoleFilter,
     RoleNestedFilter,
     RoleOrderBy,
+    SearchRoleAssignmentsInput,
     SearchRolesInput,
     UpdatePermissionInput,
     UpdateRoleInput,
@@ -93,7 +93,7 @@
     # Input models (request)
     "AdminSearchEntitiesGQLInput",
     "AdminSearchPermissionsGQLInput",
-    "AdminSearchRoleAssignmentsGQLInput",
+    "SearchRoleAssignmentsInput",
     "SearchRolesInput",
     "AssignRoleInput",
     "BulkAssignRoleInput",

src/ai/backend/common/dto/manager/v2/rbac/request.py

@@ -23,7 +23,7 @@
 __all__ = (
     "AdminSearchEntitiesGQLInput",
     "AdminSearchPermissionsGQLInput",
-    "AdminSearchRoleAssignmentsGQLInput",
+    "SearchRoleAssignmentsInput",
     "SearchRolesInput",
     "AssignRoleInput",
     "BulkAssignRoleInput",
@@ -303,8 +303,8 @@ class SearchRolesInput(BaseRequestModel):
     offset: int | None = None
 
 
-class AdminSearchRoleAssignmentsGQLInput(BaseRequestModel):
-    """GQL pagination search input for role assignments."""
+class SearchRoleAssignmentsInput(BaseRequestModel):
+    """Pagination search input for role assignments."""
 
     filter: RoleAssignmentFilter | None = None
     order: list[RoleAssignmentOrderBy] | None = None

src/ai/backend/common/dto/manager/v2/rbac/response.py

@@ -21,7 +21,7 @@
 __all__ = (
     "AdminSearchAssociationsPayload",
     "AdminSearchPermissionsPayload",
-    "AdminSearchRoleAssignmentsPayload",
+    "SearchRoleAssignmentsPayload",
     "AdminSearchRolesPayload",
     "AssociationScopesEntitiesNode",
     "BulkAssignRoleFailureInfo",
@@ -183,7 +183,7 @@ class AdminSearchPermissionsPayload(BaseResponseModel):
     has_previous_page: bool = Field(description="Whether there is a previous page.")
 
 
-class AdminSearchRoleAssignmentsPayload(BaseResponseModel):
+class SearchRoleAssignmentsPayload(BaseResponseModel):
     """Paginated result for role assignment search."""
 
     items: list[RoleAssignmentNode] = Field(description="List of role assignment nodes.")

src/ai/backend/manager/api/adapters/rbac.py

@@ -10,6 +10,7 @@
 from uuid import UUID
 
 from ai.backend.common.api_handlers import SENTINEL
+from ai.backend.common.contexts.user import current_user
 from ai.backend.common.data.filter_specs import StringMatchSpec
 from ai.backend.common.data.permission.types import OperationType as InternalOperationType
 from ai.backend.common.data.permission.types import RBACElementType
@@ -51,7 +52,7 @@
 from ai.backend.common.dto.manager.v2.rbac.request import (
     AdminSearchEntitiesGQLInput,
     AdminSearchPermissionsGQLInput,
-    AdminSearchRoleAssignmentsGQLInput,
+    SearchRoleAssignmentsInput,
     SearchRolesInput,
 )
 from ai.backend.common.dto.manager.v2.rbac.request import (
@@ -111,6 +112,7 @@
 from ai.backend.common.dto.manager.v2.rbac.types import (
     OrderDirection as OrderDirectionV2,
 )
+from ai.backend.common.exception import UnreachableError
 from ai.backend.manager.actions.action import build_operation_description
 from ai.backend.manager.api.adapters.pagination import PaginationSpec
 from ai.backend.manager.data.common.types import SearchResult
@@ -680,9 +682,22 @@ async def search_roles_in_scope(
             has_previous_page=raw.has_previous_page,
         )
 
-    async def admin_search_role_assignments_gql(
+    async def my_search_role_assignments(
         self,
-        input: AdminSearchRoleAssignmentsGQLInput,
+        input: SearchRoleAssignmentsInput,
+    ) -> SearchResult[RoleAssignmentNode]:
+        """Search role assignments for the current authenticated user."""
+        me = current_user()
+        if me is None:
+            raise UnreachableError("User context is not available")
+        return await self.admin_search_role_assignments(
+            input,
+            base_conditions=[AssignedUserConditions.by_user_id(me.user_id)],
+        )
+
+    async def admin_search_role_assignments(
+        self,
+        input: SearchRoleAssignmentsInput,
         base_conditions: Sequence[QueryCondition] | None = None,
     ) -> SearchResult[RoleAssignmentNode]:
         """Search role assignments with cursor/offset pagination."""

src/ai/backend/manager/api/gql/rbac/resolver/role.py

@@ -10,7 +10,7 @@
 from ai.backend.common.contexts.user import current_user
 from ai.backend.common.data.permission.types import RBACElementType
 from ai.backend.common.dto.manager.v2.rbac.request import (
-    AdminSearchRoleAssignmentsGQLInput,
+    SearchRoleAssignmentsInput,
     SearchRolesInput,
 )
 from ai.backend.manager.api.gql.base import encode_cursor
@@ -125,8 +125,8 @@ async def admin_role_assignments(
     offset: int | None = None,
 ) -> RoleAssignmentConnection:
     check_admin_only()
-    result = await info.context.adapters.rbac.admin_search_role_assignments_gql(
-        AdminSearchRoleAssignmentsGQLInput(
+    result = await info.context.adapters.rbac.admin_search_role_assignments(
+        SearchRoleAssignmentsInput(
             filter=filter.to_pydantic() if filter is not None else None,
             order=[o.to_pydantic() for o in order_by] if order_by is not None else None,
             first=first,
@@ -178,8 +178,8 @@ async def my_roles(
 
         raise InsufficientPrivilege("Authentication required")
 
-    result = await info.context.adapters.rbac.admin_search_role_assignments_gql(
-        AdminSearchRoleAssignmentsGQLInput(
+    result = await info.context.adapters.rbac.admin_search_role_assignments(
+        SearchRoleAssignmentsInput(
             filter=filter.to_pydantic() if filter is not None else None,
             order=[o.to_pydantic() for o in order_by] if order_by is not None else None,
             first=first,

src/ai/backend/manager/api/gql/rbac/types/role.py

@@ -16,7 +16,7 @@
 from ai.backend.common.dto.manager.v2.rbac.request import (
     AdminSearchEntitiesGQLInput,
     AdminSearchPermissionsGQLInput,
-    AdminSearchRoleAssignmentsGQLInput,
+    SearchRoleAssignmentsInput,
 )
 from ai.backend.common.dto.manager.v2.rbac.request import (
     AssignRoleInput as AssignRoleInputDTO,
@@ -281,8 +281,8 @@ async def users(
         pydantic_filter = combined_filter.to_pydantic() if combined_filter is not None else None
         pydantic_order = [o.to_pydantic() for o in order_by] if order_by is not None else None
 
-        result = await info.context.adapters.rbac.admin_search_role_assignments_gql(
-            AdminSearchRoleAssignmentsGQLInput(
+        result = await info.context.adapters.rbac.admin_search_role_assignments(
+            SearchRoleAssignmentsInput(
                 filter=pydantic_filter,
                 order=pydantic_order,
                 first=first,