Merge pull request #442 from superannotateai/assign_items

adding assign items and un_assign items functionality

Author: VaghinakDev

src/superannotate/__init__.py

@@ -2,6 +2,13 @@
 import sys
 
 sys.path.append(os.path.split(os.path.realpath(__file__))[0])
+from superannotate.lib.app.input_converters.conversion import (
+    convert_json_version,
+)  # noqa
+from superannotate.lib.app.input_converters.conversion import (
+    convert_project_type,
+)  # noqa
+
 
 import logging.config  # noqa
 import requests  # noqa

src/superannotate/lib/app/analytics/aggregators.py

@@ -1,13 +1,13 @@
 import copy
 import json
+from dataclasses import dataclass
 from pathlib import Path
 from typing import List
 from typing import Optional
 from typing import Union
 
 import lib.core as constances
 import pandas as pd
-from dataclasses import dataclass
 from lib.app.exceptions import AppException
 from lib.core import ATTACHED_VIDEO_ANNOTATION_POSTFIX
 from lib.core import PIXEL_ANNOTATION_POSTFIX

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

@@ -46,34 +46,27 @@ def __init__(
                     f"CLI's superannotate init to generate default location config file."
                 )
             config_repo = ConfigRepository(config_path)
+            main_endpoint = config_repo.get_one("main_endpoint").value
+            if not main_endpoint:
+                main_endpoint = constants.BACKEND_URL
             token, host, ssl_verify = (
                 Controller.validate_token(config_repo.get_one("token").value),
-                config_repo.get_one("main_endpoint").value,
+                main_endpoint,
                 config_repo.get_one("ssl_verify").value,
             )
         self._host = host
         self._token = token
         self.controller = Controller(token, host, ssl_verify, version)
-
-    def __new__(cls, *args, **kwargs):
-        obj = super().__new__(cls, *args, **kwargs)
-        cls.REGISTRY.append(obj)
-        return obj
+        BaseInterfaceFacade.REGISTRY.append(self)
 
     @property
-    @abstractmethod
     def host(self):
-        raise NotImplementedError
+        return self._host
 
     @property
-    @abstractmethod
     def token(self):
-        raise NotImplementedError
+        return self._token
 
-    @property
-    @abstractmethod
-    def logger(self):
-        raise NotImplementedError
 
 
 class Tracker:

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

@@ -61,6 +61,7 @@
 logger = get_default_logger()
 
 
+
 class SAClient(BaseInterfaceFacade, metaclass=TrackableMeta):
     def get_team_metadata(self):
         """Returns team metadata
@@ -312,6 +313,7 @@ def rename_project(self, project: NotEmptyStr, new_name: NotEmptyStr):
         )
         return ProjectSerializer(response.data).serialize()
 
+
     def get_folder_metadata(self, project: NotEmptyStr, folder_name: NotEmptyStr):
         """Returns folder metadata
 
@@ -667,6 +669,53 @@ def delete_images(
             f"Images deleted in project {project_name}{'/' + folder_name if folder_name else ''}"
         )
 
+    def assign_items(
+        self, project: Union[NotEmptyStr, dict], item_names: List[str], user: str
+    ):
+        """Assigns items  to a user. The assignment role, QA or Annotator, will
+        be deduced from the user's role in the project. The type of the objects` image, video or text
+        will be deduced from the project type. With SDK, the user can be
+        assigned to a role in the project with the share_project function.
+
+        :param project: project name or folder path (e.g., "project1/folder1")
+        :type project: str
+        :param item_names: list of item names to assign
+        :type item_names: list of str
+        :param user: user email
+        :type user: str
+        """
+
+        project_name, folder_name = extract_project_folder(project)
+
+        response = self.controller.assign_items(
+            project, folder_name, item_names, user
+        )
+
+        if not response.errors:
+            logger.info(f"Assign items to user {user}")
+        else:
+            raise AppException(response.errors)
+
+    def unassign_items(
+        self, project: Union[NotEmptyStr, dict], item_names: List[NotEmptyStr]
+    ):
+        """Removes assignment of given items for all assignees. With SDK,
+        the user can be assigned to a role in the project with the share_project
+        function.
+
+        :param project: project name or folder path (e.g., "project1/folder1")
+        :type project: str
+        :param item_names: list of items to unassign
+        :type item_names: list of str
+        """
+        project_name, folder_name = extract_project_folder(project)
+
+        response = self.controller.un_assign_items(
+            project_name=project_name, folder_name=folder_name, item_names=item_names
+        )
+        if response.errors:
+            raise AppException(response.errors)
+
     def assign_images(
         self, project: Union[NotEmptyStr, dict], image_names: List[str], user: str
     ):
@@ -681,6 +730,14 @@ def assign_images(
         :param user: user email
         :type user: str
         """
+
+        warning_msg = (
+            "We're deprecating the assign_images function. Please use assign_items instead."
+             "Learn more. \n"
+             "https://superannotate.readthedocs.io/en/stable/superannotate.sdk.html#superannotate.assign_items"
+        )
+        logger.warning(warning_msg)
+        warnings.warn(warning_msg, DeprecationWarning)
         project_name, folder_name = extract_project_folder(project)
         project = self.controller.get_project_metadata(project_name).data
 
@@ -719,18 +776,26 @@ def assign_images(
     def unassign_images(
         self, project: Union[NotEmptyStr, dict], image_names: List[NotEmptyStr]
     ):
-        """Removes assignment of given images for all assignees.With SDK,
+        """Removes assignment of given images for all assignees. With SDK,
         the user can be assigned to a role in the project with the share_project
         function.
 
         :param project: project name or folder path (e.g., "project1/folder1")
         :type project: str
-        :param image_names: list of image unassign
+        :param image_names: list of images to unassign
         :type image_names: list of str
         """
+
+        warning_msg = (
+            "We're deprecating the unassign_images function. Please use unassign_items instead."
+             "Learn more. \n"
+             "https://superannotate.readthedocs.io/en/stable/superannotate.sdk.html#superannotate.unassign_items"
+        )
+        logger.warning(warning_msg)
+        warnings.warn(warning_msg, DeprecationWarning)
         project_name, folder_name = extract_project_folder(project)
 
-        response = self.controller.un_assign_images(
+        response = self.controller.un_assign_items(
             project_name=project_name, folder_name=folder_name, image_names=image_names
         )
         if response.errors:

src/superannotate/lib/core/reporter.py

@@ -45,8 +45,8 @@ def __exit__(self, type, value, traceback):
 
     def __del__(self):
         globs = globals()
-        if "SESSIONS" in globs and globs["SESSIONS"].get(self.pk):
-            del globs["SESSIONS"][self.pk]
+        # if "SESSIONS" in globs and globs.get("SESSIONS", {}).get(self.pk):
+        #     del globs["SESSIONS"][self.pk]
 
     @property
     def data(self):

src/superannotate/lib/core/response.py

@@ -8,6 +8,8 @@ def __init__(self, status: str = None, data: Union[dict, list] = None):
         self._report = []
         self._errors = []
 
+    def __str__(self):
+        return f"Response object with status:{self.status}, data : {self.data}, errors: {self.errors} "
     @property
     def data(self):
         return self._data
@@ -30,7 +32,7 @@ def report_messages(self):
 
     @property
     def status(self):
-        return self.data
+        return self._status
 
     @status.setter
     def status(self, value):

src/superannotate/lib/core/serviceproviders.py

@@ -175,6 +175,16 @@ def assign_images(
     ):
         raise NotImplementedError
 
+    def assign_items(
+        self,
+        team_id: int,
+        project_id: int,
+        folder_name: str,
+        user: str,
+        item_names: list,
+    ):
+        raise NotImplementedError
+
     def get_bulk_images(
         self, project_id: int, team_id: int, folder_id: int, images: List[str]
     ) -> List[dict]:
@@ -202,6 +212,16 @@ def un_assign_images(
     ):
         raise NotImplementedError
 
+    def un_assign_items(
+        self,
+        team_id: int,
+        project_id: int,
+        folder_name: str,
+        item_names: list,
+
+    ):
+        raise NotImplementedError
+
     def un_share_project(
         self,
         team_id: int,

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

@@ -18,9 +18,13 @@
 from lib.core.repositories import BaseReadOnlyRepository
 from lib.core.response import Response
 from lib.core.serviceproviders import SuperannotateServiceProvider
+from lib.core.usecases.base import BaseUseCase
+from superannotate.logger import get_default_logger
 from lib.core.usecases.base import BaseReportableUseCase
 
 
+logger = get_default_logger()
+
 class GetItem(BaseReportableUseCase):
     def __init__(
         self,
@@ -187,6 +191,83 @@ def execute(self) -> Response:
         return self._response
 
 
+class AssignItemsUseCase(BaseUseCase):
+    CHUNK_SIZE = 500
+
+    def __init__(
+        self,
+        service: SuperannotateServiceProvider,
+        project: ProjectEntity,
+        folder: FolderEntity,
+        item_names: list,
+        user: str,
+    ):
+        super().__init__()
+        self._project = project
+        self._folder = folder
+        self._item_names = item_names
+        self._user = user
+        self._service = service
+
+    def validate_user(self, ):
+
+        for c in self._project.users:
+            if c["user_id"] == self._user:
+                return True
+
+        raise AppValidationException(f"{self._user} is not a verified contributor for the {self._project.name}")
+
+    def execute(self):
+        if self.is_valid():
+            for i in range(0, len(self._item_names), self.CHUNK_SIZE):
+                is_assigned = self._service.assign_items(
+                    team_id=self._project.team_id,
+                    project_id=self._project.id,
+                    folder_name=self._folder.name,
+                    user=self._user,
+                    item_names=self._item_names[i : i + self.CHUNK_SIZE],  # noqa: E203
+                )
+                if not is_assigned:
+                    self._response.errors = AppException(
+                        f"Cant assign {', '.join(self._item_names[i: i + self.CHUNK_SIZE])}"
+                    )
+                    continue
+        return self._response
+
+
+class UnAssignItemsUseCase(BaseUseCase):
+    CHUNK_SIZE = 500
+    def __init__(
+        self,
+        service: SuperannotateServiceProvider,
+        project_entity: ProjectEntity,
+        folder: FolderEntity,
+        item_names: list,
+    ):
+        super().__init__()
+        self._project_entity = project_entity
+        self._folder = folder
+        self._item_names = item_names
+        self._service = service
+
+    def execute(self):
+        # todo handling to backend side
+        for i in range(0, len(self._item_names), self.CHUNK_SIZE):
+            is_un_assigned = self._service.un_assign_items(
+                team_id=self._project_entity.team_id,
+                project_id=self._project_entity.id,
+                folder_name=self._folder.name,
+                item_names=self._item_names[i : i + self.CHUNK_SIZE],  # noqa: E203
+            )
+            if not is_un_assigned:
+                self._response.errors = AppException(
+                    f"Cant un assign {', '.join(self._item_names[i: i + self.CHUNK_SIZE])}"
+                )
+
+        return self._response
+
+
+
 class AttachItems(BaseReportableUseCase):
     CHUNK_SIZE = 500