Merge pull request #349 from superannotateai/friday

Friday

Author: VaghinakDev

docs/source/superannotate.sdk.rst

@@ -108,7 +108,7 @@ Team contributors
 _________________
 
 .. autofunction:: superannotate.get_team_metadata
-.. autofunction:: superannotate.invite_contributor_to_team
+.. autofunction:: superannotate.invite_contributors_to_team
 .. autofunction:: superannotate.search_team_contributors
 
 ----------

docs/source/tutorial.sdk.rst

@@ -388,7 +388,7 @@ A team contributor can be invited to the team with:
 
 .. code-block:: python
 
-   sa.invite_contributor_to_team(email="hovnatan@superannotate.com", admin=False)
+   sa.invite_contributors_to_team(emails=["hovnatan@superannotate.com"], admin=False)
 
 
 This invitation should be accepted by the contributor. After which, to share the 

src/superannotate/__init__.py

@@ -17,6 +17,7 @@
     add_annotation_comment_to_image,
 )
 from superannotate.lib.app.interface.sdk_interface import add_annotation_point_to_image
+from superannotate.lib.app.interface.sdk_interface import add_contributors_to_project
 from superannotate.lib.app.interface.sdk_interface import aggregate_annotations_as_df
 from superannotate.lib.app.interface.sdk_interface import assign_folder
 from superannotate.lib.app.interface.sdk_interface import assign_images
@@ -63,7 +64,7 @@
 from superannotate.lib.app.interface.sdk_interface import get_project_workflow
 from superannotate.lib.app.interface.sdk_interface import get_team_metadata
 from superannotate.lib.app.interface.sdk_interface import init
-from superannotate.lib.app.interface.sdk_interface import invite_contributor_to_team
+from superannotate.lib.app.interface.sdk_interface import invite_contributors_to_team
 from superannotate.lib.app.interface.sdk_interface import move_images
 from superannotate.lib.app.interface.sdk_interface import pin_image
 from superannotate.lib.app.interface.sdk_interface import prepare_export
@@ -128,7 +129,6 @@
     "convert_project_type",
     # Teams Section
     "get_team_metadata",
-    "invite_contributor_to_team",
     "search_team_contributors",
     # Projects Section
     "create_project_from_metadata",
@@ -199,6 +199,8 @@
     "consensus",
     "upload_video_to_project",
     "upload_images_to_project",
+    "add_contributors_to_project",
+    "invite_contributors_to_team",
 ]
 
 __author__ = "Superannotate"

src/superannotate/lib/app/helpers.py

@@ -87,7 +87,7 @@ def get_s3_annotation_paths(folder_path, s3_bucket, annotation_paths, recursive)
 
     paginator = s3_client.get_paginator("list_objects_v2")
     for data in paginator.paginate(Bucket=s3_bucket, Prefix=folder_path):
-        for annotation in data["Contents"]:
+        for annotation in data.get("Contents", []):
             key = annotation["Key"]
             if (
                 key.endswith(VECTOR_ANNOTATION_POSTFIX)

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

@@ -20,6 +20,7 @@
 from lib.app.helpers import get_paths_and_duplicated_from_csv
 from lib.app.interface.types import AnnotationStatuses
 from lib.app.interface.types import AnnotationType
+from lib.app.interface.types import AnnotatorRole
 from lib.app.interface.types import ImageQualityChoices
 from lib.app.interface.types import NotEmptyStr
 from lib.app.interface.types import ProjectTypes
@@ -80,19 +81,6 @@ def get_team_metadata():
     return TeamSerializer(response.data).serialize()
 
 
-@Trackable
-@validate_arguments
-def invite_contributor_to_team(email: EmailStr, admin: bool = False):
-    """Invites a contributor to team
-
-    :param email: email of the contributor
-    :type email: str
-    :param admin: enables admin priviledges for the contributor
-    :type admin: bool
-    """
-    controller.invite_contributor(email, is_admin=admin)
-
-
 @Trackable
 @validate_arguments
 def search_team_contributors(
@@ -2911,3 +2899,50 @@ def validate_annotations(
             return True
         print(response.report)
         return False
+
+
+@Trackable
+@validate_arguments
+def add_contributors_to_project(
+    project_name: NotEmptyStr, emails: List[EmailStr], role: AnnotatorRole
+) -> Tuple[List[str], List[str]]:
+    """Add contributors to project.
+
+    :param project_name: project name
+    :type project_name: str
+
+    :param emails: users email
+    :type emails: list
+
+    :param role: user role to apply, one of Admin , Annotator , QA
+    :type role: str
+
+    return: lists of added,  skipped contributors of the project
+    rtype: tuple (2 members) of lists of strs
+    """
+    response = controller.add_contributors_to_project(
+        project_name=project_name, emails=emails, role=role
+    )
+    if response.errors:
+        raise AppException(response.errors)
+    return response.data
+
+
+@Trackable
+@validate_arguments
+def invite_contributors_to_team(emails: List[EmailStr], admin: StrictBool = False) -> Tuple[List[str], List[str]]:
+    """Invites contributors to the team.
+
+    :param emails: list of contributor emails
+    :type emails: list
+
+    :param admin: enables admin privileges for the contributor
+    :type admin: bool
+
+    return: lists of invited, skipped contributors of the team
+    rtype: tuple (2 members) of lists of strs
+    """
+    response = controller.invite_contributors_to_team(emails=emails, set_admin=admin)
+    if response.errors:
+        raise AppException(response.errors)
+    return response.data

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

@@ -3,6 +3,7 @@
 
 from lib.core.enums import AnnotationStatus
 from lib.core.enums import ProjectType
+from lib.core.enums import UserRole
 from lib.core.exceptions import AppException
 from lib.infrastructure.validators import wrap_error
 from pydantic import constr
@@ -25,6 +26,20 @@ def validate(cls, value: Union[str]) -> Union[str]:
         return value
 
 
+class AnnotatorRole(StrictStr):
+    ANNOTATOR_ROLES = (UserRole.ADMIN.name, UserRole.ANNOTATOR.name, UserRole.QA.name)
+
+    @classmethod
+    def validate(cls, value: Union[str]) -> Union[str]:
+        if cls.curtail_length and len(value) > cls.curtail_length:
+            value = value[: cls.curtail_length]
+        if value.lower() not in [role.lower() for role in cls.ANNOTATOR_ROLES]:
+            raise TypeError(
+                f"Available statuses is {', '.join(AnnotatorRole)}. "
+            )
+        return value
+
+
 class AnnotationType(StrictStr):
     VALID_TYPES = ["bbox", "polygon", "point"]
 

src/superannotate/lib/app/mixp/utils/parsers.py

@@ -22,15 +22,15 @@ def get_team_metadata(*args, **kwargs):
     return {"event_name": "get_team_metadata", "properties": {}}
 
 
-def invite_contributor_to_team(*args, **kwargs):
+def invite_contributors_to_team(*args, **kwargs):
     admin = kwargs.get("admin", None)
     if not admin:
         admin = args[1:2]
     if admin:
         admin = "CUSTOM"
     else:
         admin = "DEFAULT"
-    return {"event_name": "invite_contributor_to_team", "properties": {"Admin": admin}}
+    return {"event_name": "invite_contributors_to_team", "properties": {"Admin": admin}}
 
 
 def search_team_contributors(*args, **kwargs):

src/superannotate/lib/core/enums.py

@@ -40,7 +40,7 @@ class ProjectType(BaseTitledEnum):
 
 
 class UserRole(BaseTitledEnum):
-    SUPER_AMIN = "Superadmin", 1
+    SUPER_ADMIN = "Superadmin", 1
     ADMIN = "Admin", 2
     ANNOTATOR = "Annotator", 3
     QA = "QA", 4

src/superannotate/lib/core/serviceproviders.py

@@ -42,7 +42,11 @@ def get_annotation_classes(
         raise NotImplementedError
 
     @abstractmethod
-    def share_project_bulk(self, project_id: int, team_id: int, users: list):
+    def share_project_bulk(self, project_id: int, team_id: int, users: Iterable):
+        raise NotImplementedError
+
+    @abstractmethod
+    def invite_contributors(self, team_id: int, team_role: int, emails: Iterable) -> bool:
         raise NotImplementedError
 
     @abstractmethod
@@ -57,10 +61,6 @@ def prepare_export(
     ):
         raise NotImplementedError
 
-    @abstractmethod
-    def invite_contributor(self, team_id: int, email: str, user_role: str):
-        raise NotImplementedError
-
     @abstractmethod
     def delete_team_invitation(self, team_id: int, token: str, email: str):
         raise NotImplementedError

src/superannotate/lib/core/usecases/projects.py

@@ -12,6 +12,7 @@
 from lib.core.entities import FolderEntity
 from lib.core.entities import ProjectEntity
 from lib.core.entities import ProjectSettingEntity
+from lib.core.entities import TeamEntity
 from lib.core.entities import WorkflowEntity
 from lib.core.exceptions import AppException
 from lib.core.exceptions import AppValidationException
@@ -915,31 +916,6 @@ def execute(self):
         return self._response
 
 
-class InviteContributorUseCase(BaseUseCase):
-    def __init__(
-            self,
-            backend_service_provider: SuerannotateServiceProvider,
-            email: str,
-            team_id: int,
-            is_admin: bool = False,
-    ):
-        super().__init__()
-        self._backend_service = backend_service_provider
-        self._email = email
-        self._team_id = team_id
-        self._is_admin = is_admin
-
-    def execute(self):
-        role = (
-            constances.UserRole.ADMIN.value
-            if self._is_admin
-            else constances.UserRole.ANNOTATOR.value
-        )
-        self._backend_service.invite_contributor(
-            team_id=self._team_id, email=self._email, user_role=role
-        )
-
-
 class SearchContributorsUseCase(BaseUseCase):
     def __init__(
             self,
@@ -963,3 +939,122 @@ def execute(self):
         )
         self._response.data = res
         return self._response
+
+
+class AddContributorsToProject(BaseReportableUseCae):
+    """
+    Returns tuple of lists (added, skipped)
+    """
+
+    def __init__(
+            self,
+            reporter: Reporter,
+            team: TeamEntity,
+            project: ProjectEntity,
+            emails: list,
+            role: str,
+            service: SuerannotateServiceProvider,
+    ):
+        super().__init__(reporter)
+        self._team = team
+        self._project = project
+        self._emails = emails
+        self._role = role
+        self._service = service
+
+    @property
+    def user_role(self):
+        return constances.UserRole.get_value(self._role)
+
+    def execute(self):
+        team_users = set()
+        project_users = {user["user_id"] for user in self._project.users}
+        for user in self._team.users:
+            if user.user_role > constances.UserRole.ADMIN.value:
+                team_users.add(user.email)
+        # collecting pending team users which is not admin
+        for user in self._team.pending_invitations:
+            if user["user_role"] > constances.UserRole.ADMIN.value:
+                team_users.add(user["email"])
+        # collecting pending project users which is not admin
+        for user in self._project.unverified_users:
+            if user["user_role"] > constances.UserRole.ADMIN.value:
+                project_users.add(user["email"])
+
+        to_add = team_users.intersection(self._emails) - project_users
+        to_skip = set(self._emails).difference(to_add)
+
+        if to_skip:
+            self.reporter.log_warning(
+                f"Skipped {len(to_skip)}/{len(self._emails)} "
+                "contributors that are out of the team scope or already have access to the project."
+            )
+        if to_add:
+            response = self._service.share_project_bulk(
+                team_id=self._team.uuid,
+                project_id=self._project.uuid,
+                users=[
+                    dict(user_id=user_id, user_role=self.user_role)
+                    for user_id in to_add
+                ],
+            )
+            if response and not response.get("invalidUsers"):
+                self._response.data = to_add, to_skip
+                self.reporter.log_info(
+                    f"Added {len(to_add)}/{len(self._emails)} "
+                    "contributors to the project <project_name> with the <role> role."
+                )
+        return self._response
+
+
+class InviteContributorsToTeam(BaseReportableUseCae):
+    """
+    Returns tuple of lists (added, skipped)
+    """
+
+    def __init__(
+            self,
+            reporter: Reporter,
+            team: TeamEntity,
+            emails: list,
+            set_admin: bool,
+            service: SuerannotateServiceProvider,
+    ):
+        super().__init__(reporter)
+        self._team = team
+        self._emails = emails
+        self._set_admin = set_admin
+        self._service = service
+
+    def execute(self):
+        team_users = set()
+        for user in self._team.users:
+            if user.user_role > constances.UserRole.ADMIN.value:
+                team_users.add(user.email)
+        # collecting pending team users which is not admin
+        for user in self._team.pending_invitations:
+            if user["user_role"] > constances.UserRole.ADMIN.value:
+                team_users.add(user["email"])
+
+        emails = set(self._emails)
+
+        to_skip = emails.intersection(team_users)
+        to_add = emails.difference(to_skip)
+
+        if to_skip:
+            self.reporter.log_warning(
+                f"Found {len(to_skip)}/{len(self._emails)} existing members of the team."
+            )
+        if to_add:
+            response = self._service.invite_contributors(
+                team_id=self._team.uuid,
+                # REMINDER UserRole.VIEWER is the contributor for the teams
+                team_role=constances.UserRole.ADMIN.value if self._set_admin else constances.UserRole.VIEWER.value,
+                emails=to_add
+            )
+            if response:
+                self._response.data = to_add, to_skip
+                self.reporter.log_info(
+                    f"Sent team <admin/contributor> invitations to {len(to_add)}/{len(self._emails)}."
+                )
+        return self._response