Add remove_users/remove_users_from_project methods

Author: VaghinakDev

docs/source/api_reference/api_project.rst

@@ -30,3 +30,4 @@ Projects
 .. automethod:: superannotate.SAClient.create_categories
 .. automethod:: superannotate.SAClient.list_categories
 .. automethod:: superannotate.SAClient.remove_categories
+.. automethod:: superannotate.SAClient.remove_users_from_project

docs/source/api_reference/api_team.rst

@@ -11,6 +11,7 @@ Team
 .. automethod:: superannotate.SAClient.get_user_metadata
 .. automethod:: superannotate.SAClient.set_user_custom_field
 .. automethod:: superannotate.SAClient.list_users
+.. automethod:: superannotate.SAClient.remove_users
 .. automethod:: superannotate.SAClient.pause_user_activity
 .. automethod:: superannotate.SAClient.resume_user_activity
 .. automethod:: superannotate.SAClient.get_user_scores

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

@@ -75,7 +75,6 @@
 from lib.core.jsx_conditions import EmptyQuery
 from lib.core.entities.items import ProjectCategoryEntity
 
-
 logger = logging.getLogger("sa")
 
 NotEmptyStr = constr(strict=True, min_length=1)
@@ -5337,3 +5336,69 @@ def list_workflows(self):
             EmptyQuery()
         )
         return BaseSerializer.serialize_iterable(workflows.data)
+
+    def remove_users(self, users: Union[List[int], List[str]]):
+        """
+        Allows removing users from the team.
+        :param users:   A list of emails or IDs of the users.
+        :type users: Union[List[int], List[str]]
+
+        :rtype: None:
+
+        Request Example:
+        ::
+
+            SAClient.remove_users(member=["example@gmail.com","example1@gmail.com"])
+
+        """
+        success = 0
+        if users:
+            if isinstance(users[0], int):
+                users = self.controller.work_management.list_users(id__in=users)
+                user_emails = [user.email for user in users]
+            else:
+                user_emails = users
+            if user_emails:
+                success, _ = self.controller.work_management.remove_users(user_emails)
+        logger.info(
+            f"Successfully removed {success} user(s) out of the {len(users)} provided."
+        )
+
+    def remove_users_from_project(
+        self, project: Union[NotEmptyStr, int], users: Union[List[int], List[str]]
+    ):
+        """
+        Allows removing users from the team.
+
+        :param project: The name or ID of the project.
+        :type project: Union[NotEmptyStr, int]
+
+        :param users:   A list of emails or IDs of the users.
+        :type users: Union[List[int], List[str]]
+
+        :rtype: None:
+
+        Request Example:
+        ::
+
+            SAClient.remove_users_from_project(project="Test Project", users=["example@gmail.com","example1@gmail.com"])
+
+        """
+        project = self.controller.get_project(project)
+
+        success = 0
+        if users:
+            if isinstance(users[0], int):
+                users = self.controller.work_management.list_users(
+                    project=project, id__in=users
+                )
+                user_emails = [user.email for user in users]
+            else:
+                user_emails = users
+            if user_emails:
+                success, _ = self.controller.work_management.remove_users_from_project(
+                    project, user_emails
+                )
+        logger.info(
+            f"Successfully removed {success} users(s) out of the {len(users)} provided from the project {project.name}."
+        )

src/superannotate/lib/core/serviceproviders.py

@@ -931,3 +931,11 @@ def get_custom_fields_templates(
         parent: CustomFieldEntityEnum,
     ):
         raise NotImplementedError
+
+    @abstractmethod
+    def remove_users(self, emails: List[str]):
+        raise NotImplementedError
+
+    @abstractmethod
+    def remove_users_from_project(self, project_id: int, emails: List[str]):
+        raise NotImplementedError

src/superannotate/lib/infrastructure/controller.py

@@ -177,6 +177,14 @@ def set_custom_field_value(
             context=_context,
         )
 
+    def remove_users(self, user_emails: List[str]):
+        data = self.service_provider.remove_users(user_emails)
+        return data.get("success", 0), data.get("error", 0)
+
+    def remove_users_from_project(self, project: ProjectEntity, user_emails: List[str]):
+        data = self.service_provider.remove_users_from_project(project.id, user_emails)
+        return len(data.get("succeeded", [])), len(data.get("failed", []))
+
     def list_users(
         self,
         include: List[Literal["custom_fields", "categories"]] = None,

src/superannotate/lib/infrastructure/serviceprovider.py

@@ -41,6 +41,8 @@ class ServiceProvider(BaseServiceProvider):
     URL_INVITE_CONTRIBUTORS = "api/v1/team/{}/inviteUsers"
     URL_ANNOTATION_UPLOAD_PATH_TOKEN = "images/getAnnotationsPathsAndTokens"
     URL_CREATE_WORKFLOW = "api/v1/workflows/submit"
+    URL_REMOVE_USERS_FROM_TEAM = "team/{team_id}/members/bulk"
+    URL_REMOVE_USERS_FROM_PROJECT = "project/{project_id}/share/bulk"
 
     def __init__(self, client: HttpClient):
         self.enum_mapping = {"approval_status": ApprovalStatus.get_mapping()}
@@ -339,3 +341,21 @@ def create_custom_workflow(self, org_id: str, data: dict):
             },
             data=data,
         )
+
+    def remove_users(self, emails: List[str]):
+        response = self.client.request(
+            url=self.URL_REMOVE_USERS_FROM_TEAM.format(team_id=self.client.team_id),
+            method="delete",
+            data={"data": [{"email": email} for email in emails]},
+        )
+        response.raise_for_status()
+        return response.data
+
+    def remove_users_from_project(self, project_id: int, emails: List[str]):
+        response = self.client.request(
+            url=self.URL_REMOVE_USERS_FROM_PROJECT.format(project_id=project_id),
+            method="delete",
+            data={"user_id": emails},
+        )
+        response.raise_for_status()
+        return response.data