new assign - initial

Author: dshabin

superannotate/__init__.py

@@ -59,7 +59,8 @@ def consensus(*args, **kwargs):
 )
 from .db.project_images import (
     assign_images, copy_image, copy_images, delete_images, move_image,
-    move_images, pin_image, upload_image_to_project
+    move_images, pin_image, upload_image_to_project, assign_folder,
+    unassign_folder, unassign_images
 )
 from .db.projects import (
     clone_project, create_project, create_project_from_metadata, delete_project,

superannotate/db/project_images.py

@@ -16,7 +16,8 @@
 )
 from .project_api import get_project_and_folder_metadata
 from .projects import (
-    get_project_default_image_quality_in_editor, _get_available_image_counts
+    get_project_default_image_quality_in_editor, _get_available_image_counts,
+    get_project_metadata
 )
 from ..mixp.decorators import Trackable
 from .utils import _get_upload_auth_token, _get_boto_session_by_credentials, upload_image_array_to_s3, get_image_array_to_upload, __create_image, __copy_images, __move_images, get_project_folder_string
@@ -563,30 +564,168 @@ def assign_images(project, image_names, user):
     :param user: user email
     :type user: str
     """
-    logger.info("Assign %s images to user %s", len(image_names), user)
-    if len(image_names) == 0:
-        return
-    project, project_folder = get_project_and_folder_metadata(project)
-    images = search_images((project, project_folder), return_metadata=True)
-    image_dict = {}
-    for image in images:
-        image_dict[image["name"]] = image["id"]
 
-    image_ids = []
-    for image_name in image_names:
-        image_ids.append(image_dict[image_name])
-    team_id, project_id = project["team_id"], project["id"]
-    params = {"team_id": team_id, "project_id": project_id}
-    if project_folder is not None:
-        params['folder_id'] = project_folder['id']
-    json_req = {"user_id": user, "image_ids": image_ids}
+    # TODO:
+    # logger.info("Assign %s images to user %s", len(image_names), user)
+    # if len(image_names) == 0:
+    #     return
+
+    project, folder = get_project_and_folder_metadata(project)
+    if not folder:
+        folder = 'root'
+
+    project_meta = get_project_metadata(project)
+    params = {
+        "project_id": project_meta['id'],
+        "team_id": project_meta["team_id"]
+    }
+    json_req = {
+        "image_names": image_names,
+        "assign_user_id": user,
+        "folder_name": folder,
+    }
     response = _api.send_request(
-        req_type='POST',
-        path='/images/assign',
+        req_type='PUT',
+        path='/images/editAssignment',
         params=params,
         json_req=json_req
     )
+
     if not response.ok:
         raise SABaseException(
             response.status_code, "Couldn't assign images " + response.text
         )
+
+    # images = search_images((project, project_folder), return_metadata=True)
+    # image_dict = {}
+    # for image in images:
+    #     image_dict[image["name"]] = image["id"]
+
+    # image_ids = []
+    # for image_name in image_names:
+    #     image_ids.append(image_dict[image_name])
+    # team_id, project_id = project["team_id"], project["id"]
+    # params = {"team_id": team_id, "project_id": project_id}
+    # if project_folder is not None:
+    #     params['folder_id'] = project_folder['id']
+    # json_req = {"user_id": user, "image_ids": image_ids}
+    # response = _api.send_request(
+    #     req_type='POST',
+    #     path='/images/assign',
+    #     params=params,
+    #     json_req=json_req
+    # )
+    # if not response.ok:
+    #     raise SABaseException(
+    #         response.status_code, "Couldn't assign images " + response.text
+    #     )
+
+
+def assign_folder(project, folder_name, users):
+    """Assigns folder to users. With SDK, the user can be
+    assigned to a role in the project with the share_project function.
+
+    :param project: project name or metadata of the project
+    :type project: str or dict
+    :param folder_name: folder name to assign
+    :type folder_name: str
+    :param users: list of user emails
+    :type user: list of str
+    """
+
+    project_meta = get_project_metadata(project, include_contributors=True)
+    project_users = project_meta["contributors"]
+    project_name = project_meta['name']
+    project_users = [i['user_id'] for i in project_users]
+    verified_contributor = []
+
+    for user in users:
+        if user not in project_users:
+            logging.warn(
+                f'Skipping {user} from assignees. {user} is not a verified contributor for the {project_name}'
+            )
+            continue
+        verified_contributor.append(user)
+
+    params = {
+        "project_id": project_meta['id'],
+        "team_id": project_meta["team_id"]
+    }
+    json_req = {
+        "assign_user_ids": verified_contributor,
+        "folder_name": folder_name
+    }
+    response = _api.send_request(
+        req_type='POST',
+        path='/folder/editAssignment',
+        params=params,
+        json_req=json_req
+    )
+
+    if not response.ok:
+        raise SABaseException(
+            response.status_code, "Couldn't assign folder " + response.text
+        )
+    logger.info(f'Assigned {folder_name} to users: {verified_contributor}')
+
+
+def unassign_folder(project, folder_name):
+    """Removes assignment of given folder 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 folder_name: folder name to remove assignees
+    :type folder_name: str
+    """
+
+    project_meta = get_project_metadata(project)
+    params = {
+        "project_id": project_meta['id'],
+        "team_id": project_meta["team_id"]
+    }
+    json_req = {"folder_name": folder_name, "remove_user_ids": ["all"]}
+    response = _api.send_request(
+        req_type='POST',
+        path='/folder/editAssignment',
+        params=params,
+        json_req=json_req
+    )
+
+    if not response.ok:
+        raise SABaseException(
+            response.status_code, "Couldn't unassign folder " + response.text
+        )
+    print('unassign_folder>>>>>>', response.text)
+
+
+def unassign_images(project, image_names):
+    """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
+    :type image_names: list of str
+    """
+    project_meta = get_project_metadata(project)
+    params = {
+        "project_id": project_meta['id'],
+        "team_id": project_meta["team_id"]
+    }
+    json_req = {"image_names": image_names, "remove_user_ids": ["all"]}
+    response = _api.send_request(
+        req_type='PUT',
+        path='/images/editAssignment',
+        params=params,
+        json_req=json_req
+    )
+
+    if not response.ok:
+        raise SABaseException(
+            response.status_code, "Couldn't unassign images " + response.text
+        )
+
+    print('unassign_images>>>>>>', response.text)