Add filter by project user role

src/superannotate/lib/app/interface/sdk_interface.py

@@ -562,6 +562,11 @@ def list_users(
             - email__starts: str
             - email__ends: str
 
+            Following params if project is selected::
+
+            - role: str
+            - role__in: List[str]
+
             Following params if project is not selected::
 
             - state: Literal[“Confirmed”, “Pending”]

src/superannotate/lib/core/entities/filters.py

@@ -40,14 +40,22 @@ class ProjectFilters(BaseFilters):
     status__notin: List[Literal["NotStarted", "InProgress", "Completed", "OnHold"]]
 
 
-class UserFilters(TypedDict, total=False):
+class BaseUserFilters(TypedDict, total=False):
     id: Optional[int]
     id__in: Optional[List[int]]
     email: Optional[str]
     email__in: Optional[List[str]]
     email__contains: Optional[str]
     email__starts: Optional[str]
     email__ends: Optional[str]
+
+
+class ProjectUserFilters(BaseUserFilters, total=False):
+    role: Optional[str]
+    role__in: Optional[List[str]]
+
+
+class TeamUserFilters(BaseUserFilters, total=False):
     state: Optional[str]
     state__in: Optional[List[str]]
     role: Optional[str]

src/superannotate/lib/infrastructure/controller.py

@@ -33,7 +33,8 @@
 from lib.core.entities.classes import AnnotationClassEntity
 from lib.core.entities.filters import ItemFilters
 from lib.core.entities.filters import ProjectFilters
-from lib.core.entities.filters import UserFilters
+from lib.core.entities.filters import ProjectUserFilters
+from lib.core.entities.filters import TeamUserFilters
 from lib.core.entities.integrations import IntegrationEntity
 from lib.core.entities.items import ProjectCategoryEntity
 from lib.core.entities.work_managament import ScoreEntity
@@ -56,8 +57,10 @@
 from lib.infrastructure.query_builder import IncludeHandler
 from lib.infrastructure.query_builder import ItemFilterHandler
 from lib.infrastructure.query_builder import ProjectFilterHandler
+from lib.infrastructure.query_builder import ProjectUserRoleFilterHandler
 from lib.infrastructure.query_builder import QueryBuilderChain
-from lib.infrastructure.query_builder import UserFilterHandler
+from lib.infrastructure.query_builder import TeamUserRoleFilterHandler
+from lib.infrastructure.query_builder import TeamUserStateFilterHandler
 from lib.infrastructure.repositories import S3Repository
 from lib.infrastructure.serviceprovider import ServiceProvider
 from lib.infrastructure.services.http_client import HttpClient
@@ -205,27 +208,50 @@ def list_users(
         if project:
             parent_entity = CustomFieldEntityEnum.PROJECT
             project_id = context["project_id"] = project.id
+            valid_fields = generate_schema(
+                ProjectUserFilters.__annotations__,
+                self.service_provider.get_custom_fields_templates(
+                    context, CustomFieldEntityEnum.CONTRIBUTOR, parent=parent_entity
+                ),
+            )
+            chain = QueryBuilderChain(
+                [
+                    FieldValidationHandler(valid_fields.keys()),
+                    ProjectUserRoleFilterHandler(
+                        team_id=self.service_provider.client.team_id,
+                        project=project,
+                        service_provider=self.service_provider,
+                        entity=CustomFieldEntityEnum.CONTRIBUTOR,
+                        parent=parent_entity,
+                    ),
+                ]
+            )
         else:
             parent_entity = CustomFieldEntityEnum.TEAM
             project_id = None
-        valid_fields = generate_schema(
-            UserFilters.__annotations__,
-            self.service_provider.get_custom_fields_templates(
-                context, CustomFieldEntityEnum.CONTRIBUTOR, parent=parent_entity
-            ),
-        )
-        chain = QueryBuilderChain(
-            [
-                FieldValidationHandler(valid_fields.keys()),
-                UserFilterHandler(
-                    team_id=self.service_provider.client.team_id,
-                    project_id=project_id,
-                    service_provider=self.service_provider,
-                    entity=CustomFieldEntityEnum.CONTRIBUTOR,
-                    parent=parent_entity,
+            valid_fields = generate_schema(
+                TeamUserFilters.__annotations__,
+                self.service_provider.get_custom_fields_templates(
+                    context, CustomFieldEntityEnum.CONTRIBUTOR, parent=parent_entity
                 ),
-            ]
-        )
+            )
+            chain = QueryBuilderChain(
+                [
+                    FieldValidationHandler(valid_fields.keys()),
+                    TeamUserRoleFilterHandler(
+                        team_id=self.service_provider.client.team_id,
+                        service_provider=self.service_provider,
+                        entity=CustomFieldEntityEnum.CONTRIBUTOR,
+                        parent=parent_entity,
+                    ),
+                    TeamUserStateFilterHandler(
+                        team_id=self.service_provider.client.team_id,
+                        service_provider=self.service_provider,
+                        entity=CustomFieldEntityEnum.CONTRIBUTOR,
+                        parent=parent_entity,
+                    ),
+                ]
+            )
         query = chain.handle(filters, EmptyQuery())
 
         if project and include and "categories" in include:

src/superannotate/lib/infrastructure/query_builder.py

@@ -164,17 +164,20 @@ class BaseCustomFieldHandler(AbstractQueryHandler):
     def __init__(
         self,
         team_id: int,
-        project_id: Optional[int],
         service_provider: BaseServiceProvider,
         entity: CustomFieldEntityEnum,
         parent: CustomFieldEntityEnum,
+        project: Optional[ProjectEntity] = None,
     ):
         self._service_provider = service_provider
         self._entity = entity
         self._parent = parent
         self._team_id = team_id
-        self._project_id = project_id
-        self._context = {"team_id": self._team_id, "project_id": self._project_id}
+        self._project = project
+        self._context = {
+            "team_id": self._team_id,
+            "project_id": project.id if project else None,
+        }
 
     def _handle_custom_field_key(self, key) -> Tuple[str, str, Optional[str]]:
         for custom_field in sorted(
@@ -261,7 +264,7 @@ def handle(self, filters: Dict[str, Any], query: Query = None) -> Query:
         return super().handle(filters, query)
 
 
-class UserFilterHandler(BaseCustomFieldHandler):
+class TeamUserRoleFilterHandler(BaseCustomFieldHandler):
     def _handle_special_fields(self, keys: List[str], val):
         """
         Handle special fields like 'custom_fields__'.
@@ -276,7 +279,36 @@ def _handle_special_fields(self, keys: List[str], val):
                     raise AppException("Invalid user role provided.")
             except (KeyError, AttributeError):
                 raise AppException("Invalid user role provided.")
-        elif keys[0] == "state":
+        return super()._handle_special_fields(keys, val)
+
+
+class ProjectUserRoleFilterHandler(BaseCustomFieldHandler):
+    def _handle_special_fields(self, keys: List[str], val):
+        """
+        Handle special fields like 'custom_fields__'.
+        """
+        if keys[0] == "role":
+            try:
+                if isinstance(val, list):
+                    val = [
+                        self._service_provider.get_role_id(self._project, i)
+                        for i in val
+                    ]
+                elif isinstance(val, str):
+                    val = self._service_provider.get_role_id(self._project, val)
+                else:
+                    raise AppException("Invalid user role provided.")
+            except (KeyError, AttributeError):
+                raise AppException("Invalid user role provided.")
+        return super()._handle_special_fields(keys, val)
+
+
+class TeamUserStateFilterHandler(BaseCustomFieldHandler):
+    def _handle_special_fields(self, keys: List[str], val):
+        """
+        Handle special fields like 'custom_fields__'.
+        """
+        if keys[0] == "state":
             try:
                 if isinstance(val, list):
                     val = [WMUserStateEnum[i].value for i in val]

tests/integration/work_management/test_list_users.py

@@ -33,6 +33,16 @@ def test_pending_users(self):
 
         assert project["contributors"][1]["state"] == "PENDING"
 
+    @pytest.mark.skip(reason="For not send real email")
+    def test_project_role_filter_users(self):
+        test_email = "test1@superannotate.com"
+        sa.invite_contributors_to_team(emails=[test_email])
+        sa.add_contributors_to_project(self.PROJECT_NAME, [test_email], "Annotator")
+        users = sa.list_users(project=self.PROJECT_NAME, role="QA")
+        assert len(users) == 0
+        users = sa.list_users(project=self.PROJECT_NAME, role="Annotator")
+        assert len(users) == 2
+
     def test_list_users_by_project_name(self):
         project_users = sa.list_users(project=self.PROJECT_NAME)
         assert len(project_users) == 1