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

docs/manager/graphql-reference/supergraph.graphql

@@ -14857,7 +14857,7 @@ type RoleAssignmentEdge
 input RoleAssignmentFilter
   @join__type(graph: STRAWBERRY)
 {
-  roleId: UUID = null
+  roleId: UUIDFilter = null
   role: RoleAssignmentRoleNestedFilter = null
   permission: PermissionNestedFilter = null
   username: StringFilter = null

docs/manager/graphql-reference/v2-schema.graphql

@@ -9809,7 +9809,7 @@ type RoleAssignmentEdge {
 
 """Added in 26.3.0. Filter for role assignments"""
 input RoleAssignmentFilter {
-  roleId: UUID = null
+  roleId: UUIDFilter = null
   role: RoleAssignmentRoleNestedFilter = null
   permission: PermissionNestedFilter = null
   username: StringFilter = null

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
+from uuid import UUID
+
+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=click.UUID, 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: UUID | None,
+    order_by: tuple[str, ...],
+) -> None:
+    """Search my role assignments."""
+    from ai.backend.common.dto.manager.query import UUIDFilter
+    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=UUIDFilter(equals=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

@@ -28,31 +28,31 @@ def assignment() -> None:
     multiple=True,
     help="Order by field:direction (e.g., username:asc, granted_at:desc).",
 )
-@click.option("--role-id", type=str, default=None, help="Filter by role UUID.")
+@click.option("--role-id", type=click.UUID, default=None, help="Filter by role UUID.")
 @click.option("--username-contains", type=str, default=None, help="Filter by username (contains).")
 @click.option("--email-contains", type=str, default=None, help="Filter by email (contains).")
 def search(
     limit: int | None,
     offset: int | None,
     order_by: tuple[str, ...],
-    role_id: str | None,
+    role_id: UUID | None,
     username_contains: str | None,
     email_contains: str | None,
 ) -> None:
     """Search role assignments."""
-    from ai.backend.common.dto.manager.query import StringFilter
+    from ai.backend.common.dto.manager.query import StringFilter, UUIDFilter
     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
 
     # Build filter only if any filter option is provided
     filter_dto: RoleAssignmentFilter | None = None
     if any([role_id is not None, username_contains is not None, email_contains is not None]):
         filter_dto = RoleAssignmentFilter(
-            role_id=UUID(role_id) if role_id is not None else None,
+            role_id=UUIDFilter(equals=role_id) if role_id is not None else None,
             username=(
                 StringFilter(contains=username_contains) if username_contains is not None else None
             ),
@@ -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

@@ -9,7 +9,7 @@
 from pydantic import Field, field_validator
 
 from ai.backend.common.api_handlers import SENTINEL, BaseRequestModel, Sentinel
-from ai.backend.common.dto.manager.query import DateTimeFilter, StringFilter
+from ai.backend.common.dto.manager.query import DateTimeFilter, StringFilter, UUIDFilter
 
 from .types import (
     OrderDirection,
@@ -23,7 +23,7 @@
 __all__ = (
     "AdminSearchEntitiesGQLInput",
     "AdminSearchPermissionsGQLInput",
-    "AdminSearchRoleAssignmentsGQLInput",
+    "SearchRoleAssignmentsInput",
     "SearchRolesInput",
     "AssignRoleInput",
     "BulkAssignRoleInput",
@@ -208,7 +208,7 @@ class PermissionNestedFilter(BaseRequestModel):
 class RoleAssignmentFilter(BaseRequestModel):
     """Filter for role assignments."""
 
-    role_id: UUID | None = None
+    role_id: UUIDFilter | None = None
     role: RoleNestedFilter | None = None
     permission: PermissionNestedFilter | None = None
     username: StringFilter | None = None
@@ -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,12 +682,33 @@ 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._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."""
+        """Search role assignments with cursor/offset pagination (admin)."""
+        return await self._search_role_assignments(input, base_conditions=base_conditions)
+
+    async def _search_role_assignments(
+        self,
+        input: SearchRoleAssignmentsInput,
+        base_conditions: Sequence[QueryCondition] | None = None,
+    ) -> SearchResult[RoleAssignmentNode]:
+        """Internal implementation for searching role assignments."""
         conditions = self._convert_assignment_filter(input.filter) if input.filter else []
         orders = self._convert_assignment_orders(input.order) if input.order else []
         querier = self._build_querier(
@@ -1242,7 +1265,13 @@ def _convert_permission_nested_filter(
     def _convert_assignment_filter(self, f: RoleAssignmentFilterDTO) -> list[QueryCondition]:
         conditions: list[QueryCondition] = []
         if f.role_id is not None:
-            conditions.append(AssignedUserConditions.by_role_id(f.role_id))
+            condition = self.convert_uuid_filter(
+                f.role_id,
+                equals_factory=AssignedUserConditions.by_role_id_equals,
+                in_factory=AssignedUserConditions.by_role_id_in,
+            )
+            if condition is not None:
+                conditions.append(condition)
         if f.role is not None:
             conditions.extend(self._convert_role_nested_filter(f.role))
         if f.permission is not None: