sdk_interface.py

import copy
import io
import json
import logging
import os
import sys
import uuid
from pathlib import Path
from typing import Callable
from typing import Dict
from typing import Iterable
from typing import List
from typing import Optional
from typing import Tuple
from typing import TypeVar
from typing import Union

from typing_extensions import Literal

if sys.version_info < (3, 11):
    from typing_extensions import TypedDict, NotRequired, Required  # noqa
else:
    from typing import TypedDict, NotRequired, Required  # noqa

import boto3

from tqdm import tqdm
from superannotate_core.core.conditions import CONDITION_EQ as EQ
from superannotate_core.infrastructure.repositories.item_repository import Attachment
from superannotate_core.core.conditions import Condition
from superannotate_core.core.conditions import EmptyCondition
from superannotate_core.core.enums import FolderStatus
from superannotate_core.core.enums import ClassTypeEnum
from superannotate_core.core.enums import AnnotationStatus
from superannotate_core.core.enums import ProjectType
from superannotate_core.core.enums import ApprovalStatus
from superannotate_core.core.entities import AttributeGroupSchema
from superannotate_core.core.entities import AnnotationClassEntity
from superannotate_core.core.exceptions import SAException
import lib.core as constants
from lib.app.helpers import get_annotation_paths
from lib.app.helpers import get_name_url_duplicated_from_csv
from lib.app.helpers import wrap_error as wrap_validation_errors
from lib.app.interface.base_interface import BaseInterfaceFacade
from lib.app.interface.base_interface import TrackableMeta
from lib.app.interface.types import EmailStr
from lib.app.serializers import BaseSerializer
from lib.app.serializers import ProjectSerializer
from lib.app.serializers import SettingsSerializer
from lib.app.serializers import TeamSerializer
from lib.core import LIMITED_FUNCTIONS
from lib.core import entities

from lib.core.entities import WorkflowEntity
from lib.core.entities import SettingEntity
from lib.core.entities.integrations import IntegrationEntity
from lib.core.entities.integrations import IntegrationTypeEnum
from lib.core.enums import ImageQuality
from lib.core.exceptions import AppException
from lib.core.types import MLModel

from lib.core.pydantic_v1 import constr
from lib.core.pydantic_v1 import conlist
from lib.core.pydantic_v1 import parse_obj_as
from lib.infrastructure.utils import extract_project_folder

from superannotate_core.core.entities import BaseItemEntity
from superannotate_core.app import Project, Folder


def serialize_item(entity: BaseItemEntity, project: Project, folder: Folder = None):
    if project.upload_state != constants.UploadState.EXTERNAL:
        entity.url = None
    if project.type in constants.ProjectType.images:
        if project.type == constants.ProjectType.VECTOR:
            entity.segmentation_status = None
        if project.upload_state == constants.UploadState.EXTERNAL:
            entity.prediction_status = None
            entity.segmentation_status = None
    if folder:
        entity.path = (
            f"{project.name}{f'/{folder.name}' if folder.name != 'root' else ''}"
        )
    return entity


logger = logging.getLogger("sa")

NotEmptyStr = TypeVar("NotEmptyStr", bound=constr(strict=True, min_length=1))

PROJECT_STATUS = Literal["NotStarted", "InProgress", "Completed", "OnHold"]

PROJECT_TYPE = Literal[
    "Vector",
    "Pixel",
    "Video",
    "Document",
    "Tiled",
    "PointCloud",
    "GenAI",
]

ANNOTATION_STATUS = Literal[
    "NotStarted", "InProgress", "QualityCheck", "Returned", "Completed", "Skipped"
]

APPROVAL_STATUS = Literal["Approved", "Disapproved", None]

IMAGE_QUALITY = Literal["compressed", "original"]

ANNOTATION_TYPE = Literal["bbox", "polygon", "point", "tag"]

ANNOTATOR_ROLE = Literal["Admin", "Annotator", "QA"]

FOLDER_STATUS = Literal["NotStarted", "InProgress", "Completed", "OnHold"]


class Setting(TypedDict):
    attribute: str
    value: Union[str, float, int]


class PriorityScore(TypedDict):
    name: str
    priority: float


class SAClient(BaseInterfaceFacade, metaclass=TrackableMeta):
    """Create SAClient instance to authorize SDK in a team scope.
    In case of no argument has been provided, SA_TOKEN environmental variable
    will be checked or $HOME/.superannotate/config.json will be used.

    :param token: team token
    :type token: str

    :param config_path: path to config file
    :type config_path: path-like (str or Path)

    """

    def __init__(
        self,
        token: str = None,
        config_path: str = None,
    ):
        super().__init__(token, config_path)

    def get_project_by_id(self, project_id: int):

        """Returns the project metadata

        :param project_id: the id of the project
        :type project_id: int

        :return: project metadata
        :rtype: dict
        """
        # response = self.controller.get_project_by_id(project_id=project_id)
        #
        # return ProjectSerializer(response.data).serialize()
        return Project.get_by_id(self.session, project_id).dict()

    def get_folder_by_id(self, project_id: int, folder_id: int):
        """Returns the folder metadata

        :param project_id: the id of the project
        :type project_id: int

        :param folder_id: the id of the folder
        :type folder_id: int

        :return: folder metadata
        :rtype: dict
        """
        #
        # response = self.controller.get_folder_by_id(
        #     folder_id=folder_id, project_id=project_id
        # )
        # response.raise_for_status()
        # return FolderSerializer(response.data).serialize(
        #     exclude={"completedCount", "is_root"}
        # )
        return Folder.get_by_id(self.session, project_id, folder_id).dict()

    def get_item_by_id(self, project_id: int, item_id: int):
        """Returns the item metadata

        :param project_id: the id of the project
        :type project_id: int

        :param item_id: the id of the item
        :type item_id: int

        :return: item metadata
        :rtype: dict
        """

        project = self.controller.get_project(pk=project_id)
        return serialize_item(project.get_item(pk=item_id), project).dict(
            exclude={"url", "meta"}
        )

    def get_team_metadata(self):
        """Returns team metadata

        :return: team metadata
        :rtype: dict
        """
        response = self.controller.get_team()
        return TeamSerializer(response.data).serialize()

    def search_team_contributors(
        self,
        email: EmailStr = None,
        first_name: NotEmptyStr = None,
        last_name: NotEmptyStr = None,
        return_metadata: bool = True,
    ):
        """Search for contributors in the team

        :param email: filter by email
        :type email: str

        :param first_name: filter by first name
        :type first_name: str

        :param last_name: filter by last name
        :type last_name: str

        :param return_metadata: return metadata of contributors instead of names
        :type return_metadata: bool

        :return: metadata of found users
        :rtype: list of dicts
        """

        contributors = self.controller.search_team_contributors(
            email=email, first_name=first_name, last_name=last_name
        ).data

        if not return_metadata:
            return [contributor["email"] for contributor in contributors]
        return contributors

    def search_projects(
        self,
        name: Optional[NotEmptyStr] = None,
        return_metadata: bool = False,
        include_complete_item_count: bool = False,
        status: Optional[Union[PROJECT_STATUS, List[PROJECT_STATUS]]] = None,
    ):
        """
        Project name based case-insensitive search for projects.
        If **name** is None, all the projects will be returned.

        :param name: search string
        :type name: str

        :param return_metadata: return metadata of projects instead of names
        :type return_metadata: bool

        :param include_complete_item_count: return projects that have completed items and include
            the number of completed items in response.
        :type include_complete_item_count: bool

        :param status: search projects via project status
        :type status: str

        :return: project names or metadatas
        :rtype: list of strs or dicts
        """
        statuses = []
        if status:
            if isinstance(status, (list, tuple, set)):
                statuses = list(status)
            else:
                statuses = [status]

        condition = Condition.get_empty_condition()
        if name:
            condition &= Condition("name", name, EQ)
        if include_complete_item_count:
            condition &= Condition(
                "completeImagesCount", include_complete_item_count, EQ
            )
        for status in statuses:
            condition &= Condition(
                "status", constants.ProjectStatus.get_value(status), EQ
            )

        response = self.controller.projects.list(condition)
        if response.errors:
            raise AppException(response.errors)
        if return_metadata:
            return [
                i.dict(
                    {
                        "settings",
                        "workflows",
                        "contributors",
                        "classes",
                        "item_count",
                    }
                )
                for i in response.data
            ]
        else:
            return [project.name for project in response.data]

    def create_project(
        self,
        project_name: NotEmptyStr,
        project_description: NotEmptyStr,
        project_type: PROJECT_TYPE,
        settings: List[Setting] = None,
        # fix validation for class
        # classes: List[AnnotationClassEntity] = None,
        classes: List = None,
        workflows: List = None,
        instructions_link: str = None,
    ):
        """Create a new project in the team.

        :param project_name: the new project's name
        :type project_name: str

        :param project_description: the new project's description
        :type project_description: str

        :param project_type: the new project type, Vector, Pixel, Video, Document, Tiled, PointCloud, GenAI.
        :type project_type: str

        :param settings: list of settings objects
        :type settings: list of dicts

        :param classes: list of class objects
        :type classes: list of dicts

        :param workflows:  list of information for each step
        :type workflows: list of dicts

        :param instructions_link: str of instructions URL
        :type instructions_link: str

        :return: dict object metadata the new project
        :rtype: dict
        """
        if workflows:
            if project_type.capitalize() not in (
                constants.ProjectType.VECTOR.name,
                constants.ProjectType.PIXEL.name,
            ):
                raise AppException(
                    f"Workflow is not supported in {project_type} project."
                )
            parse_obj_as(List[WorkflowEntity], workflows)
        if workflows and not classes:
            raise AppException(
                "Project with workflows can not be created without classes."
            )
        if settings:
            settings = parse_obj_as(List[SettingEntity], settings)
        else:
            settings = []
        if classes:
            classes = parse_obj_as(List[AnnotationClassEntity], classes)
        if workflows and classes:
            invalid_classes = []
            class_names = [_class.name for _class in classes]
            for step in workflows:
                if step["className"] not in class_names:
                    invalid_classes.append(step["className"])
            if invalid_classes:
                seen = set()
                seen_add = seen.add
                invalid_classes = [
                    i for i in invalid_classes if not (i in seen or seen_add(i))
                ]
                raise AppException(
                    f"There are no [{', '.join(invalid_classes)}] classes created in the project."
                )
        project_response = self.controller.projects.create(
            entities.ProjectEntity(
                name=project_name,
                description=project_description,
                type=constants.ProjectType.get_value(project_type),
                settings=settings,
                instructions_link=instructions_link,
            )
        )
        project_response.raise_for_status()
        project = project_response.data
        if classes:
            classes_response = self.controller.annotation_classes.create_multiple(
                project, classes
            )
            classes_response.raise_for_status()
            project.classes = classes_response.data
            if workflows:
                workflow_response = self.controller.projects.set_workflows(
                    project, workflows
                )
                workflow_response.raise_for_status()
                project.workflows = self.controller.projects.list_workflow(project).data
        return ProjectSerializer(project).serialize()

    # TODO fix
    def clone_project(
        self,
        project_name: Union[NotEmptyStr, dict],
        from_project: Union[NotEmptyStr, dict],
        project_description: Optional[NotEmptyStr] = None,
        copy_annotation_classes: Optional[bool] = True,
        copy_settings: Optional[bool] = True,
        copy_workflow: Optional[bool] = False,
        copy_contributors: Optional[bool] = False,
    ):
        """Create a new project in the team using annotation classes and settings from from_project.

        :param project_name: new project's name
        :type project_name: str

        :param from_project: the name of the project being used for duplication
        :type from_project: str

        :param project_description: the new project's description. If None, from_project's
                                    description will be used
        :type project_description: str

        :param copy_annotation_classes: enables copying annotation classes
        :type copy_annotation_classes: bool

        :param copy_settings: enables copying project settings
        :type copy_settings: bool

        :param copy_workflow: enables copying project workflow
        :type copy_workflow: bool

        :param copy_contributors: enables copying project contributors
        :type copy_contributors: bool

        :return: dict object metadata of the new project
        :rtype: dict
        """
        response = self.controller.projects.get_metadata(
            self.controller.get_project(from_project),
            include_annotation_classes=copy_annotation_classes,
            include_settings=copy_settings,
            include_workflow=copy_workflow,
            include_contributors=copy_contributors,
        )
        response.raise_for_status()
        project: entities.ProjectEntity = response.data
        if copy_workflow and project.type not in (
            constants.ProjectType.VECTOR,
            constants.ProjectType.PIXEL,
        ):
            raise AppException(
                f"Workflow is not supported in {project.type.name} project."
            )
        project_copy = copy.copy(project)
        if project_copy.type in (
            constants.ProjectType.VECTOR,
            constants.ProjectType.PIXEL,
        ):
            project_copy.upload_state = constants.UploadState.INITIAL
        if project_description:
            project_copy.description = project_description
        else:
            project_copy.description = project.description
        project_copy.name = project_name
        create_response = self.controller.projects.create(project_copy)
        create_response.raise_for_status()
        new_project = create_response.data
        if copy_contributors:
            logger.info(f"Cloning contributors from {from_project} to {project_name}.")
            self.controller.projects.add_contributors(
                self.controller.team, new_project, project.contributors
            )
        if copy_annotation_classes:
            logger.info(
                f"Cloning annotation classes from {from_project} to {project_name}."
            )
            classes_response = self.controller.annotation_classes.create_multiple(
                new_project, project.classes
            )
            classes_response.raise_for_status()
            project.classes = classes_response.data
        if copy_workflow:
            if not copy_annotation_classes:
                logger.info(
                    f"Skipping the workflow clone from {from_project} to {project_name}."
                )
            else:
                logger.info(f"Cloning workflow from {from_project} to {project_name}.")
                workflow_response = self.controller.projects.set_workflows(
                    new_project, project.workflows
                )
                workflow_response.raise_for_status()
                project.workflows = self.controller.projects.list_workflow(project).data
        response = self.controller.projects.get_metadata(
            new_project,
            include_settings=copy_settings,
            include_workflow=copy_workflow,
            include_contributors=copy_contributors,
            include_annotation_classes=copy_annotation_classes,
            include_complete_image_count=True,
        )

        if response.errors:
            raise AppException(response.errors)
        return ProjectSerializer(response.data).serialize()

    def create_folder(self, project: NotEmptyStr, folder_name: NotEmptyStr):
        """Create a new folder in the project.

        :param project: project name
        :type project: str

        :param folder_name: the new folder's name
        :type folder_name: str

        :return: dict object metadata the new folder
        :rtype: dict
        """

        project = self.controller.get_project(project)

        folder = project.create_folder(folder_name)
        return folder.dict(exclude={"completedCount", "is_root"})

        # if res.data:
        #     folder = res.data
        #     logger.info(f"Folder {folder.name} created in project {project.name}")
        #     return FolderSerializer(folder).serialize(
        #         exclude={"completedCount", "is_root"}
        #     )
        # if res.errors:
        #     raise AppException(res.errors)

    def delete_project(self, project: Union[NotEmptyStr, dict]):
        """Deletes the project

        :param project: project name
        :type project: str
        """
        name = project
        if isinstance(project, dict):
            name = project["name"]
        self.controller.projects.delete(name=name)

    def rename_project(self, project: NotEmptyStr, new_name: NotEmptyStr):
        """Renames the project

        :param project: project name
        :type project: str

        :param new_name: project's new name
        :type new_name: str
        """
        old_name = project
        project = self.controller.get_project(old_name)  # noqa
        project.name = new_name
        response = self.controller.projects.update(project)
        if response.errors:
            raise AppException(response.errors)
        logger.info(
            "Successfully renamed project %s to %s.", old_name, response.data.name
        )
        return ProjectSerializer(response.data).serialize()

    def get_folder_metadata(self, project: NotEmptyStr, folder_name: NotEmptyStr):
        """Returns folder metadata

        :param project: project name
        :type project: str

        :param folder_name: folder's name
        :type folder_name: str

        :return: metadata of folder
        :rtype: dict
        """
        project, folder = self.controller.get_project_folder(project, folder_name)
        if not folder:
            raise AppException("Folder not found.")
        return folder.dict(exclude={"completedCount", "is_root"})

    def delete_folders(self, project: NotEmptyStr, folder_names: List[NotEmptyStr]):
        """Delete folder in project.

        :param project: project name
        :type project: str

        :param folder_names: to be deleted folders' names
        :type folder_names: list of strs
        """
        project = self.controller.get_project(project)
        deleted_folders = project.delete_folders(folder_names)
        if deleted_folders == 0:
            raise AppException("There is no folder to delete.")
        logger.info(f"Folders {folder_names} deleted in project {project.name}")

    def search_folders(
        self,
        project: NotEmptyStr,
        folder_name: Optional[NotEmptyStr] = None,
        status: Optional[Union[FOLDER_STATUS, List[FOLDER_STATUS]]] = None,
        return_metadata: Optional[bool] = False,
    ):
        """Folder name based case-insensitive search for folders in project.

        :param project: project name
        :type project: str

        :param folder_name: the new folder's name
        :type folder_name: str. If  None, all the folders in the project will be returned.

        :param status:  search folders via status. If None, all folders will be returned. \n
            Available statuses are::

                     * NotStarted
                     * InProgress
                     * Completed
                     * OnHold
        :type status: str or list of str

        :param return_metadata: return metadata of folders instead of names
        :type return_metadata: bool

        :return: folder names or metadatas
        :rtype: list of strs or dicts

        """

        project = self.controller.get_project(project)
        condition = EmptyCondition()
        if folder_name:
            condition &= Condition("name", folder_name, EQ)
        if return_metadata:
            condition &= Condition("includeUsers", return_metadata, EQ)
        if status:
            condition &= Condition(
                "status", constants.FolderStatus.get_value(status), EQ
            )
        folders = project.list_folders(condition)
        if return_metadata:
            return [
                folder.dict(exclude={"completedCount", "is_root"})
                for folder in folders
                if not folder.is_root
            ]
        return [folder.name for folder in folders if not folder.is_root]

    def get_project_metadata(
        self,
        project: Union[NotEmptyStr, dict],
        include_annotation_classes: Optional[bool] = False,
        include_settings: Optional[bool] = False,
        include_workflow: Optional[bool] = False,
        include_contributors: Optional[bool] = False,
        include_complete_item_count: Optional[bool] = False,
    ):
        """Returns project metadata

        :param project: project name
        :type project: str

        :param include_annotation_classes: enables project annotation classes output under
                                           the key "annotation_classes"
        :type include_annotation_classes: bool

        :param include_settings: enables project settings output under
                                 the key "settings"
        :type include_settings: bool

        :param include_workflow: enables project workflow output under
                                 the key "workflow"
        :type include_workflow: bool

        :param include_contributors: enables project contributors output under
                                 the key "contributors"
        :type include_contributors: bool

        :param include_complete_item_count: enables project complete item count output under
                                 the key "completed_items_count"
        :type include_complete_item_count: bool

        :return: metadata of project
        :rtype: dict
        """
        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        response = self.controller.projects.get_metadata(
            project,
            include_annotation_classes,
            include_settings,
            include_workflow,
            include_contributors,
            include_complete_item_count,
        )
        if response.errors:
            raise AppException(response.errors)
        return ProjectSerializer(response.data).serialize()

    def get_project_settings(self, project: Union[NotEmptyStr, dict]):
        """Gets project's settings.

        Return value example: [{ "attribute" : "Brightness", "value" : 10, ...},...]

        :param project: project name or metadata
        :type project: str or dict

        :return: project settings
        :rtype: list of dicts
        """
        project_name, _ = extract_project_folder(project)
        project = self.controller.projects.get_by_name(project_name).data
        settings = self.controller.projects.list_settings(project).data
        settings = [
            SettingsSerializer(attribute.dict()).serialize() for attribute in settings
        ]
        return settings

    def get_project_workflow(self, project: Union[str, dict]):
        """Gets project's workflow.

        Return value example: [{ "step" : <step_num>, "className" : <annotation_class>, "tool" : <tool_num>, ...},...]

        :param project: project name or metadata
        :type project: str or dict

        :return: project workflow
        :rtype: list of dicts
        """
        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        workflow = self.controller.projects.list_workflow(project)
        if workflow.errors:
            raise AppException(workflow.errors)
        return workflow.data

    def search_annotation_classes(
        self,
        project: Union[NotEmptyStr, dict, Project],
        name_contains: Optional[str] = None,
    ):
        """Searches annotation classes by name_prefix (case-insensitive)

        :param project: project name
        :type project: str

        :param name_contains:  search string. Returns those classes,
         where the given string is found anywhere within its name. If None, all annotation classes will be returned.
        :type name_contains: str

        :return: annotation classes of the project
        :rtype: list of dicts
        """
        if isinstance(project, Project):
            project = project.dict()

        project_name, _ = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        condition = (
            Condition("name", name_contains, EQ) & Condition("pattern", True, EQ)
            if name_contains
            else None
        )
        return [
            annotation_class.dict()
            for annotation_class in project.list_annotation_classes(condition)
        ]

    def set_project_status(self, project: NotEmptyStr, status: PROJECT_STATUS):
        """Set project status

        :param project: project name
        :type project: str

        :param status: status to set.

            Available statuses are::

                 * NotStarted
                 * InProgress
                 * Returned
                 * Completed
                 * OnHold

        :type status: str
        """
        project = self.controller.get_project(pk=project)
        project.status = constants.ProjectStatus.get_value(status)
        response = self.controller.projects.update(project)
        if response.errors:
            raise AppException(f"Failed to change {project.name} status.")
        logger.info(f"Successfully updated {project.name} status to {status}")

    def set_folder_status(
        self, project: NotEmptyStr, folder: NotEmptyStr, status: FOLDER_STATUS
    ):
        """Set folder status

        :param project: project name
        :type project: str

        :param folder: folder name
        :type folder: str

        :param status: status to set. \n
            Available statuses are::

                     * NotStarted
                     * InProgress
                     * Completed
                     * OnHold
        :type status: str
        """
        project = self.controller.get_project(project)
        try:
            folder = project.set_folder_status(folder, FolderStatus[status])
        except SAException as e:
            raise AppException(e.message)
        except Exception:
            raise AppException(f"Failed to change {project.name}/{folder} status.")
        logger.info(
            f"Successfully updated {project.name}/{folder.name} status to {status}"
        )

    def set_project_default_image_quality_in_editor(
        self,
        project: Union[NotEmptyStr, dict],
        image_quality_in_editor: Optional[str],
    ):
        """Sets project's default image quality in editor setting.

        :param project: project name or metadata
        :type project: str or dict
        :param image_quality_in_editor: new setting value, should be "original" or "compressed"
        :type image_quality_in_editor: str
        """
        project_name, folder_name = extract_project_folder(project)
        image_quality_in_editor = ImageQuality.get_value(image_quality_in_editor)
        project = self.controller.get_project(project_name)
        response = self.controller.projects.set_settings(
            project=project,
            settings=[{"attribute": "ImageQuality", "value": image_quality_in_editor}],
        )
        if response.errors:
            raise AppException(response.errors)
        return response.data

    def pin_image(
        self,
        project: Union[NotEmptyStr, dict],
        image_name: str,
        pin: Optional[bool] = True,
    ):
        """Pins (or unpins) image

        :param project: project name or folder path (e.g., "project1/folder1")
        :type project: str
        :param image_name: image name
        :type image_name: str
        :param pin: sets to pin if True, else unpins image
        :type pin: bool
        """
        project, folder = self.controller.get_project_folder_by_path(project)
        response = self.controller.items.get_by_name(project, folder, image_name)
        if response.errors:
            raise AppException(response.errors)
        item = response.data
        item.is_pinned = int(pin)
        self.controller.items.update(project=project, item=item)

    def delete_items(self, project: str, items: Optional[List[str]] = None):
        """Delete items in a given project.

        :param project: project name or folder path (e.g., "project1/folder1")
        :type project: str
        :param items: to be deleted items' names. If None, all the items will be deleted
        :type items: list of str
        """
        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        folder = project.get_folder(folder_name)
        folder.delete_items(item_names=items)

    def assign_items(
        self, project: Union[NotEmptyStr, dict], items: 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 items: list of items to assign
        :type items: list of str

        :param user: user email
        :type user: str
        """

        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        folder = project.get_folder(folder_name)
        folder.assign_items(user=user, item_names=items)

    def unassign_items(
        self, project: Union[NotEmptyStr, dict], items: 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 items: list of items to unassign
        :type items: list of str
        """
        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        folder = project.get_folder(folder_name)
        folder.assign_items(item_names=items)

    def unassign_folder(self, project_name: NotEmptyStr, folder_name: NotEmptyStr):
        """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_name: project name
        :type project_name: str
        :param folder_name: folder name to remove assignees
        :type folder_name: str
        """
        project = self.controller.get_project(project_name)
        folder = project.get_folder(folder_name)
        folder.unassign()

    def assign_folder(
        self,
        project_name: NotEmptyStr,
        folder_name: NotEmptyStr,
        users: List[NotEmptyStr],
    ):
        """Assigns folder to users. With SDK, the user can be
        assigned to a role in the project with the share_project function.

        :param project_name: project name or metadata of the project
        :type project_name: str or dict
        :param folder_name: folder name to assign
        :type folder_name: str
        :param users: list of user emails
        :type users: list of str
        """
        # todo check to delete the function
        project = self.controller.get_project(project_name)
        project.get_folder(folder_name).assign(users=users)

    def upload_images_from_folder_to_project(
        self,
        project: Union[NotEmptyStr, dict],
        folder_path: Union[NotEmptyStr, Path],
        extensions: Optional[
            Union[List[NotEmptyStr], Tuple[NotEmptyStr]]
        ] = constants.DEFAULT_IMAGE_EXTENSIONS,
        annotation_status="NotStarted",
        from_s3_bucket=None,
        exclude_file_patterns: Optional[
            Iterable[NotEmptyStr]
        ] = constants.DEFAULT_FILE_EXCLUDE_PATTERNS,
        recursive_subfolders: Optional[bool] = False,
        image_quality_in_editor: Optional[str] = None,
    ):
        """Uploads all images with given extensions from folder_path to the project.
        Sets status of all the uploaded images to set_status if it is not None.

        If an image with existing name already exists in the project it won't be uploaded,
        and its path will be appended to the third member of return value of this
        function.

        :param project: project name or folder path (e.g., "project1/folder1")
        :type project: str or dict

        :param folder_path: from which folder to upload the images
        :type folder_path: Path-like (str or Path)

        :param extensions: tuple or list of filename extensions to include from folder
        :type extensions: tuple or list of strs

        :param annotation_status: value to set the annotation statuses of the uploaded images
         NotStarted InProgress QualityCheck Returned Completed Skipped
        :type annotation_status: str

        :param from_s3_bucket: AWS S3 bucket to use. If None then folder_path is in local filesystem
        :type from_s3_bucket: str

        :param exclude_file_patterns: filename patterns to exclude from uploading,
                default value is to exclude SuperAnnotate export related ["___save.png", "___fuse.png"]
        :type exclude_file_patterns: list or tuple of strs

        :param recursive_subfolders: enable recursive subfolder parsing
        :type recursive_subfolders: bool

        :param image_quality_in_editor: image quality be seen in SuperAnnotate web annotation editor.
                Can be either "compressed" or "original".
                If None then the default value in project settings will be used.
        :type image_quality_in_editor: str

        :return: uploaded, could-not-upload, existing-images filepaths
        :rtype: tuple (3 members) of list of strs
        """

        project_name, folder_name = extract_project_folder(project)
        if recursive_subfolders:
            logger.info(
                "When using recursive subfolder parsing same name images"
                " in different subfolders will overwrite each other."
            )
        if not isinstance(extensions, (list, tuple)):
            print(extensions)
            raise AppException(
                "extensions should be a list or a tuple in upload_images_from_folder_to_project"
            )
        elif len(extensions) < 1:
            return [], [], []

        if exclude_file_patterns:
            exclude_file_patterns = list(exclude_file_patterns) + list(
                constants.DEFAULT_FILE_EXCLUDE_PATTERNS
            )
            exclude_file_patterns = list(set(exclude_file_patterns))

        project_folder_name = project_name + (f"/{folder_name}" if folder_name else "")

        logger.info(
            "Uploading all images with extensions %s from %s to project %s. Excluded file patterns are: %s.",
            extensions,
            folder_path,
            project_folder_name,
            exclude_file_patterns,
        )

        use_case = self.controller.upload_images_from_folder_to_project(
            project_name=project_name,
            folder_name=folder_name,
            folder_path=folder_path,
            extensions=extensions,
            annotation_status=annotation_status,
            from_s3_bucket=from_s3_bucket,
            exclude_file_patterns=exclude_file_patterns,
            recursive_sub_folders=recursive_subfolders,
            image_quality_in_editor=image_quality_in_editor,
        )
        images_to_upload, duplicates = use_case.images_to_upload
        if len(duplicates):
            logger.warning(
                "%s already existing images found that won't be uploaded.",
                len(duplicates),
            )
        logger.info(
            "Uploading %s images to project %s.",
            len(images_to_upload),
            project_folder_name,
        )
        if not images_to_upload:
            return [], [], duplicates
        if use_case.is_valid():
            with tqdm(
                total=len(images_to_upload), desc="Uploading images"
            ) as progress_bar:
                for _ in use_case.execute():
                    progress_bar.update(1)
            return use_case.data
        raise AppException(use_case.response.errors)

    def download_image_annotations(
        self,
        project: Union[NotEmptyStr, dict],
        image_name: NotEmptyStr,
        local_dir_path: Union[str, Path],
    ):
        """Downloads annotations of the image (JSON and mask if pixel type project)
        to local_dir_path.

        :param project: project name or folder path (e.g., "project1/folder1")
        :type project: str

        :param image_name: image name
        :type image_name: str

        :param local_dir_path: local directory path to download to
        :type local_dir_path: Path-like (str or Path)

        :return: paths of downloaded annotations
        :rtype: tuple
        """

        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        folder = project.get_folder(folder_name)
        download_path = self.controller.setup_destination_dir(local_dir_path)
        res = self.controller.annotations.download_image_annotations(
            project=project,
            folder=folder,
            image_name=image_name,
            destination=download_path,
        )
        if res.errors:
            raise AppException(res.errors)
        return res.data

    def get_exports(
        self, project: NotEmptyStr, return_metadata: Optional[bool] = False
    ):
        """Get all prepared exports of the project.

        :param project: project name
        :type project: str

        :param return_metadata: return metadata of images instead of names
        :type return_metadata: bool

        :return: names or metadata objects of the all prepared exports of the project
        :rtype: list of strs or dicts
        """
        response = self.controller.get_exports(
            project_name=project, return_metadata=return_metadata
        )
        return response.data

    def prepare_export(
        self,
        project: Union[NotEmptyStr, dict],
        folder_names: Optional[List[NotEmptyStr]] = None,
        annotation_statuses: Optional[List[ANNOTATION_STATUS]] = None,
        include_fuse: Optional[bool] = False,
        only_pinned=False,
        **kwargs,
    ):
        """Prepare annotations and classes.json for export. Original and fused images for images with
        annotations can be included with include_fuse flag.

        :param project: project name
        :type project: str

        :param folder_names: names of folders to include in the export. If None, whole project will be exported
        :type folder_names: list of str

        :param annotation_statuses: images with which status to include, if None,
               ["NotStarted", "InProgress", "QualityCheck", "Returned", "Completed", "Skipped"]  will be chosen
               list elements should be one of NotStarted InProgress QualityCheck Returned Completed Skipped
        :type annotation_statuses: list of strs

        :param include_fuse: enables fuse images in the export
        :type include_fuse: bool

        :param only_pinned: enable only pinned output in export. This option disables all other types of output.
        :type only_pinned: bool

        :param kwargs: Arbitrary kwarg ``integration_name``
            can be provided which will be used as a storage to store export file

        :return: metadata object of the prepared export
        :rtype: dict
        """
        project_name, folder_name = extract_project_folder(project)
        if folder_names is None:
            folders = [folder_name] if folder_name else []
        else:
            folders = folder_names
        if not annotation_statuses:
            annotation_statuses = [
                constants.AnnotationStatus.NOT_STARTED.name,
                constants.AnnotationStatus.IN_PROGRESS.name,
                constants.AnnotationStatus.QUALITY_CHECK.name,
                constants.AnnotationStatus.RETURNED.name,
                constants.AnnotationStatus.COMPLETED.name,
                constants.AnnotationStatus.SKIPPED.name,
            ]
        integration_name = kwargs.get("integration_name")
        integration_id = None
        if integration_name:
            for integration in self.controller.integrations.list().data:
                if integration.name == integration_name:
                    integration_id = integration.id
                    break
            else:
                raise AppException("Integration not found.")
        response = self.controller.prepare_export(
            project_name=project_name,
            folder_names=folders,
            include_fuse=include_fuse,
            only_pinned=only_pinned,
            annotation_statuses=annotation_statuses,
            integration_id=integration_id,
        )
        if response.errors:
            raise AppException(response.errors)
        return response.data

    def upload_videos_from_folder_to_project(
        self,
        project: Union[NotEmptyStr, dict],
        folder_path: Union[NotEmptyStr, Path],
        extensions: Optional[
            Union[Tuple[NotEmptyStr], List[NotEmptyStr]]
        ] = constants.DEFAULT_VIDEO_EXTENSIONS,
        exclude_file_patterns: Optional[List[NotEmptyStr]] = (),
        recursive_subfolders: Optional[bool] = False,
        target_fps: Optional[int] = None,
        start_time: Optional[float] = 0.0,
        end_time: Optional[float] = None,
        annotation_status: Optional[ANNOTATION_STATUS] = "NotStarted",
        image_quality_in_editor: Optional[IMAGE_QUALITY] = None,
    ):
        """Uploads image frames from all videos with given extensions from folder_path to the project.
        Sets status of all the uploaded images to set_status if it is not None.

        :param project: project name or folder path (e.g., "project1/folder1")
        :type project: str

        :param folder_path: from which folder to upload the videos
        :type folder_path: Path-like (str or Path)

        :param extensions: tuple or list of filename extensions to include from folder
        :type extensions: tuple or list of strs

        :param exclude_file_patterns: filename patterns to exclude from uploading
        :type exclude_file_patterns: listlike of strs

        :param recursive_subfolders: enable recursive subfolder parsing
        :type recursive_subfolders: bool

        :param target_fps: how many frames per second need to extract from the video (approximate).
                           If None, all frames will be uploaded
        :type target_fps: float

        :param start_time: Time (in seconds) from which to start extracting frames
        :type start_time: float

        :param end_time: Time (in seconds) up to which to extract frames. If None up to end
        :type end_time: float

        :param annotation_status: value to set the annotation statuses of the uploaded images
            NotStarted InProgress QualityCheck Returned Completed Skipped
        :type annotation_status: str

        :param image_quality_in_editor: image quality be seen in SuperAnnotate web annotation editor.
            Can be either "compressed" or "original".
            If None then the default value in project settings will be used.
        :type image_quality_in_editor: str

        :return: uploaded and not-uploaded video frame images' filenames
        :rtype: tuple of list of strs
        """

        project_name, folder_name = extract_project_folder(project)

        video_paths = []
        for extension in extensions:
            if not recursive_subfolders:
                video_paths += list(Path(folder_path).glob(f"*.{extension.lower()}"))
                if os.name != "nt":
                    video_paths += list(
                        Path(folder_path).glob(f"*.{extension.upper()}")
                    )
            else:
                logger.warning(
                    "When using recursive subfolder parsing same name videos "
                    "in different subfolders will overwrite each other."
                )
                video_paths += list(Path(folder_path).rglob(f"*.{extension.lower()}"))
                if os.name != "nt":
                    video_paths += list(
                        Path(folder_path).rglob(f"*.{extension.upper()}")
                    )

        video_paths = [str(path) for path in video_paths]
        response = self.controller.upload_videos(
            project_name=project_name,
            folder_name=folder_name,
            paths=video_paths,
            target_fps=target_fps,
            start_time=start_time,
            exclude_file_patterns=exclude_file_patterns,
            end_time=end_time,
            annotation_status=annotation_status,
            image_quality_in_editor=image_quality_in_editor,
        )
        if response.errors:
            raise AppException(response.errors)
        return response.data

    def upload_video_to_project(
        self,
        project: Union[NotEmptyStr, dict],
        video_path: Union[NotEmptyStr, Path],
        target_fps: Optional[int] = None,
        start_time: Optional[float] = 0.0,
        end_time: Optional[float] = None,
        annotation_status: Optional[ANNOTATION_STATUS] = "NotStarted",
        image_quality_in_editor: Optional[IMAGE_QUALITY] = None,
    ):
        """Uploads image frames from video to platform. Uploaded images will have
        names "<video_name>_<frame_no>.jpg".

        :param project: project name or folder path (e.g., "project1/folder1")
        :type project: str

        :param video_path: video to upload
        :type video_path: Path-like (str or Path)

        :param target_fps: how many frames per second need to extract from the video (approximate).
                           If None, all frames will be uploaded
        :type target_fps: float

        :param start_time: Time (in seconds) from which to start extracting frames
        :type start_time: float

        :param end_time: Time (in seconds) up to which to extract frames. If None up to end
        :type end_time: float

        :param annotation_status: value to set the annotation statuses of the uploaded
                                  video frames NotStarted InProgress QualityCheck Returned Completed Skipped
        :type annotation_status: str

        :param image_quality_in_editor: image quality be seen in SuperAnnotate web annotation editor.
                Can be either "compressed" or "original".
                If None then the default value in project settings will be used.
        :type image_quality_in_editor: str

        :return: filenames of uploaded images
        :rtype: list of strs
        """

        project_name, folder_name = extract_project_folder(project)

        response = self.controller.upload_videos(
            project_name=project_name,
            folder_name=folder_name,
            paths=[video_path],
            target_fps=target_fps,
            start_time=start_time,
            end_time=end_time,
            annotation_status=annotation_status,
            image_quality_in_editor=image_quality_in_editor,
        )
        if response.errors:
            raise AppException(response.errors)
        return response.data

    def create_annotation_class(
        self,
        project: Union[NotEmptyStr, dict, Project],
        name: NotEmptyStr,
        color: NotEmptyStr,
        attribute_groups: Optional[List[AttributeGroupSchema]] = None,
        class_type: str = "object",
    ):
        """Create annotation class in project

        :param project: project name
        :type project: str

        :param name: name for the class
        :type name: str

        :param color: RGB hex color value, e.g., "#F9E0FA"
        :type color: str

        :param attribute_groups:  list of attribute group dicts.
            The values for the "group_type" key are
            ::

                * radio
                * checklist
                * checklist
                * text
                * numeric
                * ocr

            `ocr` and `group_type` keys are only available for Vector projects.
            Mandatory keys for each attribute group are::

              - "name"
        :type attribute_groups: list of dicts

        :param class_type: class type. Should be either "object" or "tag". Document project type can also have "relationship" type of classes.
        :type class_type: str

        :return: new class metadata
        :rtype: dict

        Request Example:
        ::

            attributes_list = [
                {
                    "group_type": "radio",
                    "name": "Vehicle",
                    "attributes": [
                       {
                           "name": "Car"
                       },
                       {
                           "name": "Track"
                       },
                       {
                           "name": "Bus"
                       }
                   ],
                   "default_value": "Car"
               },
               {
                   "group_type": "checklist",
                   "name": "Color",
                   "attributes": [
                       {
                           "name": "Yellow"
                       },
                       {
                           "name": "Black"
                       },
                       {
                           "name": "White"
                       }
                   ],
                   "default_value": ["Yellow", "White"]
               },
               {
                   "group_type": "text",
                   "name": "Timestamp"
               },
               {
                   "group_type": "numeric",
                   "name": "Description"
               }
            ]
            client.create_annotation_class(
               project="Image Project",
               name="Example Class",
               color="#F9E0FA",
               attribute_groups=attributes_list
            )

        """
        if isinstance(project, Project):
            project = project.dict()

        project_name, _ = extract_project_folder(project)
        project = self.controller.get_project(project_name)

        if project.type == ProjectType.Pixel.value and any(
            "default_value" in attr_group.keys() for attr_group in attribute_groups
        ):
            raise AppException(
                'The "default_value" key is not supported for project type Pixel.'
            )

        _class_type = ClassTypeEnum.get_value(class_type)
        if (
            project.type != ProjectType.Document
            and _class_type == ClassTypeEnum.relationship
        ):
            raise AppException(
                f"{class_type} class type is not supported in {project.type.name} project."
            )
        annotation_class = project.create_annotation_class(
            name=name,
            color=color,
            class_type=_class_type,
            attribute_groups=attribute_groups,
        )
        if annotation_class:
            return annotation_class.dict()
        raise AppException("Failed to create annotation class")

    def delete_annotation_class(
        self,
        project: Union[NotEmptyStr, dict, Project],
        annotation_class: Union[dict, NotEmptyStr],
    ):
        """Deletes annotation class from project

        :param project: project name
        :type project: str

        :param annotation_class: annotation class name or  metadata
        :type annotation_class: str or dict
        """

        if isinstance(annotation_class, dict) and "name" in annotation_class.keys():
            class_name = annotation_class["name"]
        elif isinstance(annotation_class, str):
            class_name = annotation_class
        else:
            raise AppException("Invalid value provided for annotation_class.")

        if isinstance(project, Project):
            project = project.dict()

        project_name, _ = extract_project_folder(project)
        project = self.controller.get_project(project_name)

        condition = Condition("name", class_name, EQ) & Condition("pattern", True, EQ)
        annotation_classes = project.list_annotation_classes(condition=condition)
        if annotation_classes:
            class_to_delete = annotation_classes[0]
            logger.info(
                "Deleting annotation class from project %s with name %s",
                project.name,
                class_to_delete.name,
            )
            project.delete_annotation_class(class_id=class_to_delete.id)

    def download_annotation_classes_json(
        self, project: Union[NotEmptyStr, dict, Project], folder: Union[str, Path]
    ):
        """Downloads project classes.json to folder

        :param project: project name
        :type project: str

        :param folder: folder to download to
        :type folder: Path-like (str or Path)

        :return: path of the download file
        :rtype: str
        """
        if isinstance(project, Project):
            project = project.dict()

        project_name, _ = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        logger.info(
            f"Downloading classes.json from project {project.name} to folder {str(folder)}."
        )
        annotation_classes: List[dict] = [
            annotation_class.dict()
            for annotation_class in project.list_annotation_classes()
        ]
        json_path = f"{folder}/classes.json"
        with open(json_path, "w") as f:
            json.dump(annotation_classes, f, indent=4)
        return json_path

    def create_annotation_classes_from_classes_json(
        self,
        project: Union[NotEmptyStr, dict, Project],
        classes_json: Union[List[AnnotationClassEntity], str, Path],
        from_s3_bucket: str = None,
    ):
        """Creates annotation classes in project from a SuperAnnotate format
        annotation classes.json.

        :param project: project name
        :type project: str

        :param classes_json: JSON itself or path to the JSON file
        :type classes_json: list or Path-like (str or Path)

        :param from_s3_bucket: AWS S3 bucket to use. If None then classes_json is in local filesystem
        :type from_s3_bucket: str

        :return: list of created annotation class metadatas
        :rtype: list of dicts
        """
        if isinstance(classes_json, (str, Path)):
            if from_s3_bucket:
                from_session = boto3.Session()
                from_s3 = from_session.resource("s3")
                file = io.BytesIO()
                from_s3_object = from_s3.Object(from_s3_bucket, classes_json)
                from_s3_object.download_fileobj(file)
                file.seek(0)
                data = file
            else:
                data = open(classes_json, encoding="utf-8")
            classes_json = json.load(data)

        if isinstance(project, Project):
            project = project.dict()

        project_name, _ = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        annotation_classes = project.create_annotation_classes(classes_json)
        return [i.dict() for i in annotation_classes]

    def download_export(
        self,
        project: Union[NotEmptyStr, dict],
        export: Union[NotEmptyStr, dict],
        folder_path: Union[str, Path],
        extract_zip_contents: Optional[bool] = True,
        to_s3_bucket=None,
    ):
        """Download prepared export.

        :param project: project name
        :type project: str

        :param export: export name
        :type export: str or dict

        :param folder_path: where to download the export
        :type folder_path: Path-like (str or Path)

        :param extract_zip_contents: if False then a zip file will be downloaded,
         if True the zip file will be extracted at folder_path
        :type extract_zip_contents: bool

        :param to_s3_bucket: AWS S3 bucket to use for download. If None then folder_path is in local filesystem.
        :type to_s3_bucket: Bucket object
        """
        project_name, folder_name = extract_project_folder(project)
        export_name = export["name"] if isinstance(export, dict) else export

        response = self.controller.download_export(
            project_name=project_name,
            export_name=export_name,
            folder_path=folder_path,
            extract_zip_contents=extract_zip_contents,
            to_s3_bucket=to_s3_bucket,
        )
        if response.errors:
            raise AppException(response.errors)

    def set_project_workflow(
        self, project: Union[NotEmptyStr, dict], new_workflow: List[dict]
    ):
        """Sets project's workflow.

        new_workflow example: [{ "step" : <step_num>, "className" : <annotation_class>, "tool" : <tool_num>,
          "attribute":[{"attribute" : {"name" : <attribute_value>, "attribute_group" : {"name": <attribute_group>}}},
          ...]},...]

        :param project: project name or metadata
        :type project: str or dict

        :param new_workflow: new workflow list of dicts
        :type new_workflow: list of dicts
        """
        project_name, _ = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        response = self.controller.projects.set_workflows(project, steps=new_workflow)
        if response.errors:
            raise AppException(response.errors)

    def download_image(
        self,
        project: Union[NotEmptyStr, dict],
        image_name: NotEmptyStr,
        local_dir_path: Optional[Union[str, Path]] = "./",
        include_annotations: Optional[bool] = False,
        include_fuse: Optional[bool] = False,
        include_overlay: Optional[bool] = False,
        variant: Optional[str] = "original",
    ):
        """Downloads the image (and annotation if not None) to local_dir_path

        :param project: project name or folder path (e.g., "project1/folder1")
        :type project: str

        :param image_name: image name
        :type image_name: str

        :param local_dir_path: where to download the image
        :type local_dir_path: Path-like (str or Path)

        :param include_annotations: enables annotation download with the image
        :type include_annotations: bool

        :param include_fuse: enables fuse image download with the image
        :type include_fuse: bool

        :param include_overlay: enables overlay image download with the image
        :type include_overlay: bool

        :param variant: which resolution to download, can be 'original' or 'lores'
         (low resolution used in web editor)
        :type variant: str

        :return: paths of downloaded image and annotations if included
        :rtype: tuple
        """
        project_name, folder_name = extract_project_folder(project)
        response = self.controller.download_image(
            project_name=project_name,
            folder_name=folder_name,
            image_name=image_name,
            download_path=str(local_dir_path),
            image_variant=variant,
            include_annotations=include_annotations,
            include_fuse=include_fuse,
            include_overlay=include_overlay,
        )
        if response.errors:
            raise AppException(response.errors)
        logger.info(f"Downloaded image {image_name} to {local_dir_path} ")
        return response.data

    def upload_annotations(
        self,
        project: Union[NotEmptyStr, dict],
        annotations: List[dict],
        keep_status: bool = False,
    ):
        """Uploads a list of annotation dicts as annotations to the SuperAnnotate directory.

        :param project: project name or folder path (e.g., "project1/folder1")
        :type project: str or dict

        :param annotations:  list of annotation dictionaries corresponding to SuperAnnotate format
        :type annotations: list of dicts

        :param keep_status: If False, the annotation status will be automatically
            updated to "InProgress," otherwise the current status will be kept.
        :type keep_status: bool


        :return: a dictionary containing lists of successfully uploaded, failed and skipped name
        :rtype: dict

        Response Example:
        ::

            {
               "succeeded": [],
               "failed":[],
               "skipped": []
            }

        """
        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        folder = project.get_folder(folder_name)

        failed, skipped = [], []
        name_annotation_map = {}
        for annotation in annotations:
            try:
                name = annotation["metadata"]["name"]
                name_annotation_map[name] = annotation
            except KeyError:
                failed.append(annotation)
        logger.info(
            f"Uploading {len(name_annotation_map)}/{len(annotations)} "
            f"annotations to the project {project.name}."
        )

        folder_items = folder.list_items(item_names=list(name_annotation_map.keys()))
        name_item_map = {i.name: i for i in folder_items}
        len_existing, len_provided = len(folder_items), len(name_annotation_map)
        if len_existing < len_provided:
            logger.warning(
                f"Couldn't find {len_provided - len_existing}/{len_provided} "
                "items in the given directory that match the annotations."
            )
        item_id_annotation_pairs: List[Tuple[int, dict]] = []
        item_id_name_map = {}
        for annotation_name, annotation in name_annotation_map.items():
            item = name_item_map.get(annotation_name)
            if item:
                # Verifies value is not NaN for data integrity
                try:
                    json.dumps(annotation, allow_nan=False)
                except ValueError:
                    failed.append(annotation_name)
                    continue

                item_id_annotation_pairs.append((item.id, annotation))
                item_id_name_map[item.id] = annotation_name
            else:
                skipped.append(annotation_name)

        failed_ids = folder.upload_annotations(item_id_annotation_pairs)
        failed.extend(item_id_name_map[i] for i in failed_ids)
        uploaded_annotations = list(
            set(item_id_name_map.values()) - set(failed).union(set(skipped))
        )
        if uploaded_annotations and not keep_status:
            try:
                folder.set_items_annotation_statuses(
                    items=uploaded_annotations,
                    annotation_status=constants.AnnotationStatus.IN_PROGRESS,
                )
            except Exception:
                raise AppException("Failed to change status.")

        return {
            "succeeded": uploaded_annotations,
            "failed": failed,
            "skipped": skipped,
        }

    def upload_annotations_from_folder_to_project(
        self,
        project: Union[NotEmptyStr, dict],
        folder_path: Union[str, Path],
        from_s3_bucket=None,
        recursive_subfolders: Optional[bool] = False,
        keep_status=False,
    ):
        """Finds and uploads all JSON files in the folder_path as annotations to the project.

        The JSON files should follow specific naming convention. For Vector
        projects they should be named "<image_filename>___objects.json" (e.g., if
        image is cats.jpg the annotation filename should be cats.jpg___objects.json), for Pixel projects
        JSON file should be named "<image_filename>___pixel.json" and also second mask
        image file should be present with the name "<image_name>___save.png". In both cases
        image with <image_name> should be already present on the platform.

        Existing annotations will be overwritten.

        :param project: project name or folder path (e.g., "project1/folder1")
        :type project: str or dict

        :param folder_path: from which folder to upload annotations
        :type folder_path: str or dict

        :param from_s3_bucket: AWS S3 bucket to use. If None then folder_path is in local filesystem
        :type from_s3_bucket: str

        :param recursive_subfolders: enable recursive subfolder parsing
        :type recursive_subfolders: bool

        :param keep_status:   If False, the annotation status will be automatically
         updated to "InProgress," otherwise the current status will be kept.
        :type keep_status: bool

        :return: paths to annotations uploaded, could-not-upload, missing-images
        :rtype: tuple of list of strs
        """

        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        folder = project.get_folder(folder_name)
        project_folder_name = project_name + (f"/{folder_name}" if folder_name else "")

        if recursive_subfolders:
            logger.info(
                "When using recursive subfolder parsing same name annotations in different "
                "subfolders will overwrite each other.",
            )
        logger.info(
            "The JSON files should follow a specific naming convention, matching file names already present "
            "on the platform. Existing annotations will be overwritten"
        )

        annotation_paths = get_annotation_paths(
            folder_path, from_s3_bucket, recursive_subfolders
        )

        logger.info(
            f"Uploading {len(annotation_paths)} annotations from {folder_path} to the project {project_folder_name}."
        )
        response = self.controller.annotations.upload_from_folder(
            project=project,
            folder=folder,
            annotation_paths=annotation_paths,  # noqa: E203
            client_s3_bucket=from_s3_bucket,
            folder_path=folder_path,
            keep_status=keep_status,
        )
        if response.errors:
            raise AppException(response.errors)
        return response.data

    def upload_image_annotations(
        self,
        project: Union[NotEmptyStr, dict],
        image_name: str,
        annotation_json: Union[str, Path, dict],
        mask: Optional[Union[str, Path, bytes]] = None,
        verbose: Optional[bool] = True,
        keep_status: bool = False,
    ):
        """Upload annotations from JSON (also mask for pixel annotations)
        to the image.

        :param project: project name or folder path (e.g., "project1/folder1")
        :type project: str

        :param image_name: image name
        :type image_name: str

        :param annotation_json: annotations in SuperAnnotate format JSON dict or path to JSON file
        :type annotation_json: dict or Path-like (str or Path)

        :param mask: BytesIO object or filepath to mask annotation for pixel projects in SuperAnnotate format
        :type mask: BytesIO or Path-like (str or Path)

        :param verbose: Turns on verbose output logging during the proces.
        :type verbose: bool

        :param keep_status:   If False, the annotation status will be automatically
         updated to "InProgress," otherwise the current status will be kept.
        :type keep_status: bool

        """

        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        folder = project.get_folder(folder_name)
        if project.type not in constants.ProjectType.images:
            raise AppException(LIMITED_FUNCTIONS[project.type])

        if not mask:
            if not isinstance(annotation_json, dict):
                mask_path = str(annotation_json).replace("___pixel.json", "___save.png")
            else:
                mask_path = f"{image_name}___save.png"
            if os.path.exists(mask_path):
                mask = open(mask_path, "rb").read()
        elif isinstance(mask, str) or isinstance(mask, Path):
            if os.path.exists(mask):
                mask = open(mask, "rb").read()

        if not isinstance(annotation_json, dict):
            if verbose:
                logger.info("Uploading annotations from %s.", annotation_json)
            annotation_json = json.load(open(annotation_json))
        if not folder:
            raise AppException("Folder not found.")

        image = self.controller.items.get_by_name(project, folder, image_name).data

        if not image:
            raise AppException("Image not found.")
        response = self.controller.annotations.upload_image_annotations(
            project=project,
            folder=folder,
            image=image,
            annotations=annotation_json,
            user=self.controller.current_user,
            mask=mask,
            verbose=verbose,
            keep_status=keep_status,
        )
        if response.errors and not response.errors == constants.INVALID_JSON_MESSAGE:
            raise AppException(response.errors)

    def download_model(self, model: MLModel, output_dir: Union[str, Path]):
        """Downloads the neural network and related files
        which are the <model_name>.pth/pkl. <model_name>.json, <model_name>.yaml, classes_mapper.json

        :param model: the model that needs to be downloaded
        :type  model: dict

        :param output_dir: the directory in which the files will be saved
        :type output_dir: str

        :return: the metadata of the model
        :rtype: dict
        """
        res = self.controller.models.download(
            model_data=model.dict(), download_path=output_dir
        )
        if res.errors:
            logger.error("\n".join([str(error) for error in res.errors]))
        else:
            return BaseSerializer(res.data).serialize()

    def consensus(
        self,
        project: NotEmptyStr,
        folder_names: List[int],
        image_list: Optional[List[NotEmptyStr]] = None,
        annotation_type: Optional[ANNOTATION_TYPE] = "bbox",
    ):
        """Computes consensus score for each instance of given images
            that are present in at least 2 of the given projects:

        :param project: project name
        :type project: str

        :param folder_names: list of folder names in the project for which the scores will be computed
        :type folder_names: list of str

        :param image_list: List of image names from the projects list that must be used.
                If None, then all images from the projects list will be used. Default: None
        :type image_list: list

        :param annotation_type: Type of annotation instances to consider.
                Available candidates are: ["bbox", "polygon", "point"]
        :type annotation_type: str

        :return: Pandas DateFrame with columns
                (creatorEmail, QA, imageName, instanceId, className, area, attribute, folderName, score)
        :rtype: pandas DataFrame
        """

        response = self.controller.consensus(
            project_name=project,
            folder_names=folder_names,
            image_list=image_list,
            annot_type=annotation_type,
        )
        if response.errors:
            raise AppException(response.errors)
        return response.data

    def run_prediction(
        self,
        project: Union[NotEmptyStr, dict],
        images_list: List[NotEmptyStr],
        model: Union[NotEmptyStr, dict],
    ):
        """This function runs smart prediction on given list of images from a given project
            using the neural network of your choice

        :param project: the project in which the target images are uploaded.
        :type project: str or dict

        :param images_list: the list of image names on which smart prediction has to be run
        :type images_list: list of str

        :param model: the name of the model that should be used for running smart prediction
        :type model: str or dict

        :return: tuple of two lists, list of images on which the prediction has succeeded and failed respectively
        :rtype: tuple
        """
        project_name = None
        folder_name = None
        if isinstance(project, dict):
            project_name = project["name"]
        if isinstance(project, str):
            project_name, folder_name = extract_project_folder(project)

        model_name = model
        if isinstance(model, dict):
            model_name = model["name"]
        project, folder = self.controller.get_project_folder(project_name, folder_name)
        response = self.controller.models.run_prediction(
            project=project,
            folder=folder,
            items_list=images_list,
            model_name=model_name,
        )
        if response.errors:
            raise AppException(response.errors)
        return response.data

    def upload_image_to_project(
        self,
        project: NotEmptyStr,
        img,
        image_name: Optional[NotEmptyStr] = None,
        annotation_status: Optional[ANNOTATION_STATUS] = "NotStarted",
        from_s3_bucket=None,
        image_quality_in_editor: Optional[NotEmptyStr] = None,
    ):
        """Uploads image (io.BytesIO() or filepath to image) to project.
        Sets status of the uploaded image to set_status if it is not None.

        :param project: project name or folder path (e.g., "project1/folder1")
        :type project: str

        :param img: image to upload
        :type img: io.BytesIO() or Path-like (str or Path)

        :param image_name: image name to set on platform. If None and img is filepath,
                           image name will be set to filename of the path
        :type image_name: str

        :param annotation_status: value to set the annotation statuses of the uploaded image
                NotStarted InProgress QualityCheck Returned Completed Skipped
        :type annotation_status: str

        :param from_s3_bucket: AWS S3 bucket to use. If None then folder_path is in local filesystem
        :type from_s3_bucket: str

        :param image_quality_in_editor: image quality be seen in SuperAnnotate web annotation editor.
                Can be either "compressed" or "original".
                If None then the default value in project settings will be used.
        :type image_quality_in_editor: str
        """
        project_name, folder_name = extract_project_folder(project)

        response = self.controller.upload_image_to_project(
            project_name=project_name,
            folder_name=folder_name,
            image_name=image_name,
            image=img,
            annotation_status=annotation_status,
            from_s3_bucket=from_s3_bucket,
            image_quality_in_editor=image_quality_in_editor,
        )
        if response.errors:
            raise AppException(response.errors)

    def search_models(
        self,
        name: Optional[NotEmptyStr] = None,
        type_: Optional[NotEmptyStr] = None,  # noqa
        project_id: Optional[int] = None,
        task: Optional[NotEmptyStr] = None,
        include_global: Optional[bool] = True,
    ):
        r"""Search for ML models.

        :param name: search string
        :type name: str

        :param type\_: ml model type string
        :type type\_: str

        :param project_id: project id
        :type project_id: int

        :param task: training task
        :type task: str

        :param include_global: include global ml models
        :type include_global: bool

        :return: ml model metadata
        :rtype: list of dicts
        """
        condition = EmptyCondition()
        if name:
            condition &= Condition("name", name, EQ)
        if type_:
            condition &= Condition("type", type_, EQ)
        if project_id:
            condition &= Condition("project_id", project_id, EQ)
        if task:
            condition &= Condition("task", task, EQ)
        if include_global:
            condition &= Condition("include_global", include_global, EQ)

        res = self.controller.models.list(condition)
        return res.data

    def upload_images_to_project(
        self,
        project: NotEmptyStr,
        img_paths: List[NotEmptyStr],
        annotation_status: Optional[ANNOTATION_STATUS] = "NotStarted",
        from_s3_bucket=None,
        image_quality_in_editor: Optional[IMAGE_QUALITY] = None,
    ):
        """Uploads all images given in list of path objects in img_paths to the project.
        Sets status of all the uploaded images to set_status if it is not None.

        If an image with existing name already exists in the project it won't be uploaded,
        and its path will be appended to the third member of return value of this
        function.

        :param project: project name or folder path (e.g., "project1/folder1")
        :type project: str

        :param img_paths: list of Path-like (str or Path) objects to upload
        :type img_paths: list

        :param annotation_status: value to set the annotation statuses of the uploaded images
            Available statuses are::

                     * NotStarted
                     * InProgress
                     * QualityCheck
                     * Returned
                     * Completed
                     * Skipped
        :type annotation_status: str

        :param from_s3_bucket: AWS S3 bucket to use. If None then folder_path is in local filesystem
        :type from_s3_bucket: str

        :param image_quality_in_editor: image quality be seen in SuperAnnotate web annotation editor.
                Can be either "compressed" or "original".
                If None then the default value in project settings will be used.
        :type image_quality_in_editor: str

        :return: uploaded, could-not-upload, existing-images filepaths
        :rtype: tuple (3 members) of list of strs
        """
        project_name, folder_name = extract_project_folder(project)

        use_case = self.controller.upload_images_to_project(
            project_name=project_name,
            folder_name=folder_name,
            paths=img_paths,
            annotation_status=annotation_status,
            image_quality_in_editor=image_quality_in_editor,
            from_s3_bucket=from_s3_bucket,
        )

        images_to_upload, existing_items = use_case.images_to_upload
        logger.info(f"Uploading {len(images_to_upload)} images to project {project}.")
        uploaded, failed_images = [], []
        if not images_to_upload:
            return uploaded, failed_images, existing_items
        if use_case.is_valid():
            with tqdm(
                total=len(images_to_upload), desc="Uploading images"
            ) as progress_bar:
                for _ in use_case.execute():
                    progress_bar.update(1)
            uploaded, failed_images, existing_items = use_case.data
            if existing_items:
                logger.info(f"Existing images {', '.join(existing_items)}")
            return uploaded, failed_images, existing_items
        raise AppException(use_case.response.errors)

    def aggregate_annotations_as_df(
        self,
        project_root: Union[NotEmptyStr, Path],
        project_type: PROJECT_TYPE,
        folder_names: Optional[List[Union[Path, NotEmptyStr]]] = None,
    ):
        """Aggregate annotations as pandas dataframe from project root.

        :param project_root: the export path of the project
        :type project_root: Path-like (str or Path)

        :param project_type: the project type, Vector/Pixel, Video or Document
        :type project_type: str

        :param folder_names: Aggregate the specified folders from project_root.
         If None aggregates all folders in the project_root
        :type folder_names: list of Pathlike (str or Path) objects

        :return: DataFrame on annotations
        :rtype: pandas DataFrame
        """
        from superannotate.lib.app.analytics.aggregators import DataAggregator

        return DataAggregator(
            project_type=project_type,  # noqa
            project_root=project_root,
            folder_names=folder_names,
        ).aggregate_annotations_as_df()

    def delete_annotations(
        self, project: NotEmptyStr, item_names: Optional[List[NotEmptyStr]] = None
    ):
        """
        Delete item annotations from a given list of items.

        :param project: project name or folder path (e.g., "project1/folder1")
        :type project: str
        :param item_names:  item names. If None, all the annotations in the specified directory will be deleted.
        :type item_names: list of strs
        """

        project, folder = self.controller.get_project_folder_by_path(project)
        response = self.controller.annotations.delete(
            project=project, folder=folder, item_names=item_names
        )
        if response.errors:
            raise AppException(response.errors)

    def validate_annotations(
        self,
        project_type: PROJECT_TYPE,
        annotations_json: Union[NotEmptyStr, Path, dict],
    ):
        """Validates given annotation JSON.

        :param project_type: The project type Vector, Pixel, Video or Document
        :type project_type: str

        :param annotations_json: path to annotation JSON
        :type annotations_json: Path-like (str or Path)

        :return: The success of the validation
        :rtype: bool
        """
        if isinstance(annotations_json, dict):
            annotation_data = annotations_json
        else:
            annotation_data = json.load(open(annotations_json))
        response = self.controller.validate_annotations(project_type, annotation_data)
        if response.errors:
            raise AppException(response.errors)
        report = response.data
        if not report:
            return True
        print(wrap_validation_errors(report))
        return False

    def add_contributors_to_project(
        self,
        project: NotEmptyStr,
        emails: conlist(EmailStr, min_items=1),
        role: ANNOTATOR_ROLE,
    ) -> Tuple[List[str], List[str]]:
        """Add contributors to project.

        :param project: project name
        :type project: 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
        """
        project = self.controller.projects.get_by_name(project).data
        contributors = [
            entities.ContributorEntity(
                user_id=email, user_role=constants.UserRole(role)
            )
            for email in emails
        ]
        response = self.controller.projects.add_contributors(
            team=self.controller.get_team().data,
            project=project,
            contributors=contributors,
        )
        if response.errors:
            raise AppException(response.errors)
        return response.data

    def invite_contributors_to_team(
        self, emails: conlist(EmailStr, min_items=1), admin: bool = 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 = self.controller.invite_contributors_to_team(
            emails=emails, set_admin=admin
        )
        if response.errors:
            raise AppException(response.errors)
        return response.data

    def get_annotations(
        self,
        project: Union[NotEmptyStr, int],
        items: Optional[Union[List[NotEmptyStr], List[int]]] = None,
    ):
        """Returns annotations for the given list of items.

        :param project: project id or project name or folder path (e.g., “project1/folder1”).
        :type project: str or int

        :param items:  item names. If None, all the items in the specified directory will be used.
        :type items: list of strs or list of ints

        :return: list of annotations
        :rtype: list of dict
        """
        folder = None
        if isinstance(items, list) and not items:
            return []
        names_provided = all([isinstance(i, str) for i in items]) if items else False
        ids_provided = all([isinstance(i, int) for i in items]) if items else False
        if items is not None and not (names_provided or ids_provided):
            raise AppException("Provide an int list or a str list.")
        if isinstance(project, str):
            project_name, folder_name = extract_project_folder(project)
            project = self.controller.get_project(project_name)
            if folder_name:
                folder = project.get_folder(folder_name)
        else:
            project = self.controller.get_project(project)
        if not folder:
            folder = project.get_folder("root")
        if names_provided:
            return folder.get_annotations(item_names=items)
        else:
            return folder.get_annotations(item_ids=items)

    def get_annotations_per_frame(
        self, project: NotEmptyStr, video: NotEmptyStr, fps: int = 1
    ):
        """Returns per frame annotations for the given video.


        :param project: project name or folder path (e.g., “project1/folder1”).
        :type project: str

        :param video: video name
        :type video: str

        :param fps: how many frames per second needs to be extracted from the video.
         Will extract 1 frame per second by default.
        :type fps: str

        :return: list of annotation objects
        :rtype: list of dicts
        """
        project_name, folder_name = extract_project_folder(project)
        response = self.controller.get_annotations_per_frame(
            project_name, folder_name, video_name=video, fps=fps
        )
        if response.errors:
            raise AppException(response.errors)
        return response.data

    def upload_priority_scores(
        self,
        project: Union[NotEmptyStr, dict],
        scores: Union[List[PriorityScore], List[dict[str, float]]],
    ):
        """Upload priority scores for the given list of items.

        :param project: project name or folder path (e.g., “project1/folder1”)
        :type project: str

        :param scores: list of score objects
        :type scores: list of dicts

        :return: lists of uploaded, skipped items
        :rtype: tuple (2 members) of lists of strs
        """
        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        folder = project.get_folder(folder_name)
        logger.info(
            f"Uploading  priority scores for {len(scores)} item(s) to {folder.name}."
        )
        uploaded_score_names = folder.upload_priority_scores(scores)
        skipped_score_names = list(
            {i["name"] for i in scores} - set(uploaded_score_names)
        )
        return uploaded_score_names, skipped_score_names

    def get_integrations(self):
        """Get all integrations per team

        :return: metadata objects of all integrations of the team.
        :rtype: list of dicts
        """
        response = self.controller.integrations.list()
        if response.errors:
            raise AppException(response.errors)
        integrations = response.data
        return BaseSerializer.serialize_iterable(
            integrations, ("name", "type", "root")  # noqa
        )

    def attach_items_from_integrated_storage(
        self,
        project: NotEmptyStr,
        integration: Union[NotEmptyStr, IntegrationEntity],
        folder_path: Optional[NotEmptyStr] = None,
    ):
        """Link images from integrated external storage to SuperAnnotate.

        :param project: project name or folder path where items should be attached (e.g., “project1/folder1”).
        :type project: str

        :param integration:  existing integration name or metadata dict to pull items from.
         Mandatory keys in integration metadata’s dict is “name”.
        :type integration: str or dict

        :param folder_path: Points to an exact folder/directory within given storage.
         If None, items     are fetched from the root directory.
        :type folder_path: str
        """
        project, folder = self.controller.get_project_folder_by_path(project)
        _integration = None
        if isinstance(integration, str):
            integration = IntegrationEntity(name=integration)
        for i in self.controller.integrations.list().data:
            if integration.name == i.name:
                _integration = i
                break
        else:
            raise AppException("Integration not found.")
        response = self.controller.integrations.attach_items(
            project, folder, _integration, folder_path
        )
        if response.errors:
            raise AppException(response.errors)

    def query(
        self,
        project: NotEmptyStr,
        query: Optional[NotEmptyStr] = None,
        subset: Optional[NotEmptyStr] = None,
    ):
        """Return items that satisfy the given query.
        Query syntax should be in SuperAnnotate query language(https://doc.superannotate.com/docs/explore-overview).

        :param project: project name or folder path (e.g., “project1/folder1”)
        :type project: str

        :param query: SAQuL query string.
        :type query: str

        :param subset:  subset name. Allows you to query items in a specific subset.
            To return all the items in the specified subset, set the value of query param to None.
        :type subset: str

        :return: queried items’ metadata list
        :rtype: list of dicts
        """
        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        subset_id = None
        if subset:
            subset = next(
                (i for i in project.list_subsets() if i["name"] == subset), None
            )
            if not subset:
                raise AppException("Subset not found")
            subset_id = subset["id"]
        if folder_name:
            folder = project.get_folder(folder_name)
            _items = folder.list_items(query=query, subset_id=subset_id)
        else:
            _items = project.list_items(query=query, subset_id=subset_id)
        exclude = {"meta"}
        return [serialize_item(i, project).dict(exclude=exclude) for i in _items]

    def get_item_metadata(
        self,
        project: NotEmptyStr,
        item_name: NotEmptyStr,
        include_custom_metadata: bool = False,
    ):
        """Returns item metadata

        :param project: project name or folder path (e.g., “project1/folder1”)
        :type project: str

        :param item_name: item name.
        :type item_name: str

        :param include_custom_metadata: include custom metadata that has been attached to an asset.
        :type include_custom_metadata: bool

        :return: metadata of item
        :rtype: dict

        Request Example:
        ::

            client.get_item_metadata(
               project="Medical Annotations",
               item_name = "image_1.png",
               include_custom_metadata=True
            )

        Response Example:
        ::

            {
               "name": "image_1.jpeg",
               "path": "Medical Annotations/Study",
               "url": "https://sa-public-files.s3.../image_1.png",
               "annotation_status": "NotStarted",
               "annotator_email": None,
               "qa_email": None,
               "entropy_value": None,
               "createdAt": "2022-02-15T20:46:44.000Z",
               "updatedAt": "2022-02-15T20:46:44.000Z",
               "custom_metadata": {
                   "study_date": "2021-12-31",
                   "patient_id": "62078f8a756ddb2ca9fc9660",
                   "patient_sex": "female",
                   "medical_specialist": "robertboxer@ms.com",
               }
            }
        """
        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        folder = project.get_folder(folder_name)
        item = folder.get_item(
            item_name, include_custom_metadata=include_custom_metadata
        )
        exclude = {"custom_metadata"} if not include_custom_metadata else set()
        exclude.add("meta")
        return serialize_item(item, project, folder).dict(exclude=exclude)

    def search_items(
        self,
        project: NotEmptyStr,
        name_contains: NotEmptyStr = None,
        annotation_status: Optional[ANNOTATION_STATUS] = None,
        annotator_email: Optional[NotEmptyStr] = None,
        qa_email: Optional[NotEmptyStr] = None,
        recursive: bool = False,
        include_custom_metadata: bool = False,
    ):
        """Search items by filtering criteria.

        :param project: project name or folder path (e.g., “project1/folder1”).
            If recursive=False=True, then only the project name is required.
        :type project: str

        :param name_contains:  Returns those items, where the given string is found anywhere within an item’s name.
            If None, all items returned, in accordance with the recursive=False parameter.
        :type name_contains: str

        :param annotation_status: if not None, filters items by annotation status. \n
            Available statuses are::

                     * NotStarted
                     * InProgress
                     * QualityCheck
                     * Returned
                     * Completed
                     * Skipped
        :type annotation_status: str

        :param annotator_email: returns those items’ names that are assigned to the specified annotator.
            If None, all items are returned. Strict equal.
        :type annotator_email: str

        :param qa_email:  returns those items’ names that are assigned to the specified QA.
            If None, all items are returned. Strict equal.
        :type qa_email: str

        :param recursive: search in the project’s root and all of its folders.
            If False search only in the project’s root or given directory.
        :type recursive: bool

        :param include_custom_metadata: include custom metadata that has been attached to an asset.
        :type include_custom_metadata: bool

        :return: metadata of item
        :rtype: list of dicts

        Request Example:
        ::

            client.search_items(
               project="Medical Annotations",
               name_contains="image_1",
               include_custom_metadata=True
            )

        Response Example:
        ::

            [
               {
                   "name": "image_1.jpeg",
                   "path": "Medical Annotations/Study",
                   "url": "https://sa-public-files.s3.../image_1.png",
                   "annotation_status": "NotStarted",
                   "annotator_email": None,
                   "qa_email": None,
                   "entropy_value": None,
                   "createdAt": "2022-02-15T20:46:44.000Z",
                   "updatedAt": "2022-02-15T20:46:44.000Z",
                   "custom_metadata": {
                       "study_date": "2021-12-31",
                       "patient_id": "62078f8a756ddb2ca9fc9660",
                       "patient_sex": "female",
                       "medical_specialist": "robertboxer@ms.com",
                   }
               }
            ]
        """
        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)

        search_condition = Condition.get_empty_condition()
        if name_contains:
            search_condition &= Condition("name", name_contains, EQ)
        if annotation_status:
            search_condition &= Condition(
                "annotation_status",
                constants.AnnotationStatus.get_value(annotation_status),
                EQ,
            )
        if qa_email:
            search_condition &= Condition("qa_id", qa_email, EQ)
        if annotator_email:
            search_condition &= Condition("annotator_id", annotator_email, EQ)

        exclude = {"custom_metadata"} if not include_custom_metadata else set()
        exclude.add("meta")
        if recursive:
            items = []
            for folder in project.list_folders():
                items.extend(
                    [
                        serialize_item(i, project, folder).dict(exclude=exclude)
                        for i in folder.list_items(
                            condition=search_condition,
                            include_custom_metadata=include_custom_metadata,
                        )
                    ]
                )
        else:
            folder = project.get_folder(folder_name)
            items = [
                serialize_item(i, project, folder).dict(exclude=exclude)
                for i in folder.list_items(
                    condition=search_condition,
                    include_custom_metadata=include_custom_metadata,
                )
            ]
        return items

    def attach_items(
        self,
        project: Union[NotEmptyStr, dict],
        attachments: Union[NotEmptyStr, Path, conlist(Attachment, min_items=1)],
        annotation_status: Optional[ANNOTATION_STATUS] = "NotStarted",
    ):
        """Link items from external storage to SuperAnnotate using URLs.

        :param project: project name or folder path (e.g., “project1/folder1”)
        :type project: str

        :param attachments: path to CSV file or list of dicts containing attachments URLs.
        :type attachments: path-like (str or Path) or list of dicts

        :param annotation_status: value to set the annotation statuses of the linked items. \n
                Available statuses are::

                 * NotStarted
                 * InProgress
                 * QualityCheck
                 * Returned
                 * Completed
                 * Skipped
        :type annotation_status: str

        :return: uploaded, failed and duplicated item names
        :rtype: tuple of list of strs

        Example:
        ::

            client = SAClient()
            client.attach_items(
                project="Medical Annotations",
                attachments=[{"name": "item", "url": "https://..."}]
             )

        Example of attaching items from custom integration:
        ::

            client = SAClient()
            client.attach_items(
                project="Medical Annotations",
                attachments=[
                    {
                        "name": "item",
                        "url": "https://bucket-name.s3…/example.png"
                        "integration": "custom-integration-name"
                        }
                    ]
            )
        """

        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        folder = project.get_folder(folder_name)
        # todo validate Attachment

        seen_names = set()
        unique_attachments = []
        if isinstance(attachments, (str, Path)):
            (
                unique_attachments,
                duplicate_attachments,
            ) = get_name_url_duplicated_from_csv(attachments)
            len_duplicate_names = len(duplicate_attachments)
        else:
            for i in attachments:
                name = i.get("name", None)
                if name is None:
                    i["name"] = str(uuid.uuid4())
                    unique_attachments.append(i)
                elif name not in seen_names:
                    seen_names.add(name)
                    unique_attachments.append(i)
            len_duplicate_names = len(attachments) - len(unique_attachments)

        if len_duplicate_names:
            logger.info("Dropping duplicates.")
        uploaded, fails, duplicated = [], [], []
        _unique_attachments = []
        if any("integration" in i for i in unique_attachments):
            integtation_item_map = {
                i.name: i
                for i in self.controller.integrations.list().data
                if i.type == IntegrationTypeEnum.CUSTOM
            }
            invalid_integrations = set()
            for attachment in unique_attachments:
                if "integration" in attachment:
                    if attachment["integration"] in integtation_item_map:
                        attachment["integration_id"] = integtation_item_map[
                            attachment["integration"]
                        ].id
                    elif attachment["integration"]:
                        invalid_integrations.add(attachment["integration"])
                        continue
                _unique_attachments.append(attachment)
            if invalid_integrations:
                logger.error(
                    f"The ['{','.join(invalid_integrations)}'] integrations specified for the items doesn't exist in the "
                    "list of integrations on the platform. Any associated items will be skipped."
                )
        else:
            _unique_attachments = unique_attachments

        if _unique_attachments:
            logger.info(
                f"Attaching {len(_unique_attachments)} file(s) to project {project.name}."
            )

            uploaded, duplicated = folder.attach_items(
                attachments=_unique_attachments,
                annotation_status=AnnotationStatus[annotation_status],
            )
            fails = [
                attachment["name"]
                for attachment in _unique_attachments
                if attachment["name"] not in uploaded
                and attachment["name"] not in duplicated
            ]
        return uploaded, fails, duplicated

    def copy_items(
        self,
        source: Union[NotEmptyStr, dict],
        destination: Union[NotEmptyStr, dict],
        items: Optional[List[NotEmptyStr]] = None,
        include_annotations: Optional[bool] = True,
    ):
        """Copy images in bulk between folders in a project

        :param source: project name or folder path to select items from (e.g., “project1/folder1”).
        :type source: str

        :param destination: project name (root) or folder path to place copied items.
        :type destination: str

        :param items: names of items to copy. If None, all items from the source directory will be copied.
        :type items: list of str

        :param include_annotations: enables annotations copy
        :type include_annotations: bool

        :return: list of skipped item names
        :rtype: list of strs
        """

        project_name, source_folder = extract_project_folder(source)
        to_project_name, destination_folder = extract_project_folder(destination)
        if project_name != to_project_name:
            raise AppException("Source and destination projects should be the same")
        project = self.controller.get_project(project_name)
        from_folder = project.get_folder(source_folder)
        to_folder = project.get_folder(destination_folder)
        skipped_items = from_folder.copy_items_by_name(
            destination_folder_id=to_folder.id,
            items=items,
            include_annotations=include_annotations,
        )
        return skipped_items

    def move_items(
        self,
        source: Union[NotEmptyStr, dict],
        destination: Union[NotEmptyStr, dict],
        items: Optional[List[NotEmptyStr]] = None,
    ):
        """Move images in bulk between folders in a project

        :param source: project name or folder path to pick items from (e.g., “project1/folder1”).
        :type source: str

        :param destination: project name (root) or folder path to move items to.
        :type destination: str

        :param items: names of items to move. If None, all items from the source directory will be moved.
        :type items: list of str

        :return: list of skipped item names
        :rtype: list of strs
        """

        project_name, source_folder = extract_project_folder(source)
        to_project_name, destination_folder = extract_project_folder(destination)
        if project_name != to_project_name:
            raise AppException("Source and destination projects should be the same")

        project = self.controller.get_project(project_name)
        source_folder = project.get_folder(source_folder)
        destination_folder = project.get_folder(destination_folder)
        skipped_names = source_folder.move_items_by_name(
            destination_folder_id=destination_folder.id, items=items
        )
        return skipped_names

    def set_annotation_statuses(
        self,
        project: Union[NotEmptyStr, dict],
        annotation_status: ANNOTATION_STATUS,
        items: Optional[List[NotEmptyStr]] = None,
    ):
        """Sets annotation statuses of items

        :param project: project name or folder path (e.g., “project1/folder1”).
        :type project: str

        :param annotation_status: annotation status to set. \n
            Available statuses are::

                 * NotStarted
                 * InProgress
                 * QualityCheck
                 * Returned
                 * Completed
                 * Skipped
        :type annotation_status: str

        :param items:  item names. If None, all the items in the specified directory will be used.
        :type items: list of strs
        """

        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        folder = project.get_folder(folder_name)
        folder.set_items_annotation_statuses(
            items=items, annotation_status=AnnotationStatus[annotation_status]
        )
        logger.info("Annotation statuses of items changed")

    def download_annotations(
        self,
        project: Union[NotEmptyStr, dict],
        path: Union[str, Path] = None,
        items: Optional[List[NotEmptyStr]] = None,
        recursive: bool = False,
        callback: Callable = None,
    ):
        """Downloads annotation JSON files of the selected items to the local directory.

        :param project: project name or folder path (e.g., “project1/folder1”).
        :type project: str

        :param path:  local directory path where the annotations will be downloaded.
                If none, the current directory is used.
        :type path: Path-like (str or Path)

        :param items: list of item names whose annotations will be downloaded
                (e.g., ["Image_1.jpeg", "Image_2.jpeg"]). If the value is None,
                then all the annotations of the given directory will be downloaded.

        :type items: list of str

        :param recursive: download annotations from the project’s root
                and all of its folders with the preserved structure.
                If False download only from the project’s root or given directory.
        :type recursive: bool

        :param callback: a function that allows you to modify each annotation’s dict before downloading.
         The function receives each annotation as an argument and the returned value will be applied to the download.
        :type callback: callable

        :return: local path of the downloaded annotations folder.
        :rtype: str
        """
        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)

        download_path = self.controller.setup_destination_dir(path)
        classes_download_path = f"{download_path}/classes"
        self.controller.download_annotation_classes(
            project_id=project.id, path=classes_download_path
        )
        if recursive:
            project.download_annotations(
                download_path=download_path, item_names=items, callback=callback
            )
        else:
            folder = project.get_folder(folder_name)
            folder.download_annotations(
                download_path=download_path, item_names=items, callback=callback
            )
        return download_path

    def get_subsets(self, project: Union[NotEmptyStr, dict]):
        """Get Subsets

        :param project: project name (e.g., “project1”)
        :type project: str

        :return: subsets’ metadata
        :rtype: list of dicts
        """
        project_name, _ = extract_project_folder(project)
        project = self.controller.projects.get_by_name(project_name).data
        response = self.controller.subsets.list(project)
        if response.errors:
            raise AppException(response.errors)
        return BaseSerializer.serialize_iterable(response.data, ["name"])

    def create_custom_fields(self, project: NotEmptyStr, fields: dict):
        """Create custom fields for items in a project in addition to built-in metadata.
        Using this function again with a different schema won't override the existing fields, but add new ones.
        Use the upload_custom_values() function to fill them with values for each item.

        :param project: project name  (e.g., “project1”)
        :type project: str

        :param fields:  dictionary describing the fields and their specifications added to the project.
         You can see the schema structure <here>.
        :type fields: dict

        :return: custom fields actual schema of the project
        :rtype: dict

        Supported Types:

        ==============  ======================
                    number
        --------------------------------------
         field spec           spec value
        ==============  ======================
        minimum         any number (int or float)
        maximum         any number (int or float)
        enum            list of numbers (int or float)
        ==============  ======================

        ==============  ======================
                    string
        --------------------------------------
         field spec           spec value
        ==============  ======================
        format          “email” (user@example.com) or “date” (YYYY-MM-DD)

        enum            list of strings
        ==============  ======================

        ::

            custom_fields = {
               "study_date": {
                   "type": "string",
                   "format": "date"
               },
               "patient_id": {
                   "type": "string"
               },
               "patient_sex": {
                   "type": "string",
                   "enum": [
                       "male", "female"
                   ]
               },
               "patient_age": {
                   "type": "number"
               },
               "medical_specialist": {
                   "type": "string",
                   "format": "email"
               },
               "duration": {
                   "type": "number",
                   "minimum": 10
               }
            }

            client = SAClient()
            client.create_custom_fields(
               project="Medical Annotations",
               fields=custom_fields
            )

        """
        project_name, _ = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        return project.create_custom_fields(fields)

    def get_custom_fields(self, project: NotEmptyStr):
        """Get the schema of the custom fields defined for the project

        :param project: project name  (e.g., “project1”)
        :type project: str

        :return: custom fields actual schema of the project
        :rtype: dict

        Response Example:
        ::

            {
               "study_date": {
                   "type": "string",
                   "format": "date"
               },
               "patient_id": {
                   "type": "string"
               },
               "patient_sex": {
                   "type": "string",
                   "enum": [
                       "male", "female"
                   ]
               },
               "patient_age": {
                   "type": "number"
               },
               "medical_specialist": {
                   "type": "string",
                   "format": "email"
               },
               "duration": {
                   "type": "number",
                   "minimum": 10
               }
            }
        """
        project_name, _ = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        return project.get_custom_fields()

    def delete_custom_fields(
        self, project: NotEmptyStr, fields: conlist(str, min_items=1)
    ):
        """Remove custom fields from a project’s custom metadata schema.

        :param project: project name  (e.g., “project1”)
        :type project: str

        :param fields: list of field names to remove
        :type fields: list of strs

        :return: custom fields actual schema of the project
        :rtype: dict

        Request Example:
        ::

            client = SAClient()
            client.delete_custom_fields(
               project = "Medical Annotations",
               fields = ["duration", patient_age]
            )

        Response Example:
        ::

            {
                "study_date": {
                   "type": "string",
                   "format": "date"
                },
                "patient_id": {
                   "type": "string"
                },
                "patient_sex": {
                   "type": "string",
                   "enum": [
                       "male", "female"
                   ]
                },
                "medical_specialist": {
                   "type": "string",
                   "format": "email"
                }
            }

        """
        project_name, _ = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        project.delete_custom_field(fields)
        return project.get_custom_fields()

    def upload_custom_values(
        self, project: NotEmptyStr, items: conlist(Dict[str, dict], min_items=1)
    ):
        """
        Attach custom metadata to items.
        SAClient.get_item_metadata(), SAClient.search_items(), SAClient.query() methods
        will return the item metadata and custom metadata.

        :param project: project name or folder path (e.g., “project1/folder1”)
        :type project: str

        :param items:  list of name-data pairs.
            The key of each dict indicates an existing item name and the value represents the custom metadata dict.
            The values for the corresponding keys will be added to an item or will be overridden.
        :type items: list of dicts

        :return: dictionary with succeeded and failed item names.
        :rtype: dict

        Request Example:
        ::

            client = SAClient()

            items_values = [
               {
                   "image_1.png": {
                       "study_date": "2021-12-31",
                       "patient_id": "62078f8a756ddb2ca9fc9660",
                       "patient_sex": "female",
                       "medical_specialist": "robertboxer@ms.com"
                   }
               },
               {
                   "image_2.png": {
                       "study_date": "2021-12-31",
                       "patient_id": "62078f8a756ddb2ca9fc9661",
                       "patient_sex": "female",
                       "medical_specialist": "robertboxer@ms.com"
                   }
               },
               {
                   "image_3.png": {
                       "study_date": "2011-10-05T14:48:00.000Z",
                       "patient_": "62078f8a756ddb2ca9fc9660",
                       "patient_sex": "female",
                       "medical_specialist": "robertboxer"
                   }
               }
            ]

            client.upload_custom_values(
               project = "Medical Annotations",
               items = items_values
            )

        Response Example:
        ::

            {
               "succeeded": ["image_1.png", "image_2.png"],
               "failed": ["image_3.png"]
            }
        """

        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        folder = project.get_folder(folder_name)
        _item_fields_mapping = {}
        for i in items:
            _item_fields_mapping.update(i)
        _items = folder.list_items(item_names=list(_item_fields_mapping.keys()))
        item_fields_map = {i: _item_fields_mapping[i.name] for i in _items}
        succeeded_items, failed_items = folder.set_custom_field_values(item_fields_map)
        return {"succeeded": succeeded_items, "failed": failed_items}

    def delete_custom_values(
        self, project: NotEmptyStr, items: conlist(Dict[str, List[str]], min_items=1)
    ):
        """
        Remove custom data from items

        :param project: project name or folder path (e.g., “project1/folder1”)
        :type project: str

        :param items:   list of name-custom data dicts.
            The key of each dict element indicates an existing item in the project root or folder.
            The value should be the list of fields to be removed from the given item.
            Please note, that the function removes pointed metadata from a given item.
            To delete metadata for all items you should delete it from the custom metadata schema.
            To override values for existing fields, use SAClient.upload_custom_values()
        :type items: list of dicts

        Request Example:
        ::

            client.delete_custom_values(
                project = "Medical Annotations",
                items = [
                   {"image_1.png": ["study_date", "patient_sex"]},
                   {"image_2.png": ["study_date", "patient_sex"]}
                ]
            )
        """
        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        folder = project.get_folder(folder_name)
        _item_fields_mapping = {}
        for i in items:
            _item_fields_mapping.update(i)
        _items = folder.list_items(item_names=list(_item_fields_mapping.keys()))
        item_fields_map = {i: _item_fields_mapping[i.name] for i in _items}
        return folder.delete_custom_field_values(item_fields_map)

    def add_items_to_subset(
        self, project: NotEmptyStr, subset: NotEmptyStr, items: List[dict]
    ):

        # todo we can update the interface because there has been no usage since may 26
        """

        Associates selected items with a given subset. Non-existing subset will be automatically created.

        :param project:  project name (e.g., “project1”)
        :type project: str

        :param subset: a name of an existing/new subset to associate items with.
                New subsets will be automatically created.
        :type subset: str

        :param items: list of items metadata.
                Required keys are 'name' and 'path' if the 'id' key is not provided in the dict.
        :type items: list of dicts

        :return: dictionary with succeeded, skipped and failed items lists.
        :rtype: dict

        Request Example:
        ::

            client = SAClient()

            # option 1
            queried_items = client.query(
                project="Image Project",
                query="instance(error = true)"
             )

            client.add_items_to_subset(
                project="Medical Annotations",
                subset="Brain Study - Disapproved",
                items=queried_items
            )
            # option 2
            items_list = [
                {
                    'name': 'image_1.jpeg',
                    'path': 'Image Project'
                },
                {
                    'name': 'image_2.jpeg',
                    'path': 'Image Project/Subfolder A'
                }
            ]

            client.add_items_to_subset(
                project="Image Project",
                subset="Subset Name",
                items=items_list

            )

        Response Example:
        ::

            {
                "succeeded": [
                    {
                        'name': 'image_1.jpeg',
                        'path': 'Image Project'
                    },
                    {
                        'name': 'image_2.jpeg',
                        'path': 'Image Project/Subfolder A'
                    }
                ],
                "failed": [],
                "skipped": []
            }
        """

        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        item_ids = [i["id"] for i in items]
        successed, skipped, failed = project.add_items_to_subset(subset, item_ids)
        return {"successed": successed, "skipped": skipped, "failed": failed}

    def set_approval_statuses(
        self,
        project: NotEmptyStr,
        approval_status: Optional[APPROVAL_STATUS],
        items: Optional[List[NotEmptyStr]] = None,
    ):
        """Sets annotation statuses of items

        :param project: project name or folder path (e.g., “project1/folder1”).
        :type project: str

        :param approval_status: approval status to set. \n
            Available statuses are::

                     * None
                     * Approved
                     * Disapproved
        :type approval_status: str

        :param items:  item names to set the mentioned status for. If None, all the items in the project will be used.
        :type items: list of strs
        """
        project_name, folder_name = extract_project_folder(project)
        project = self.controller.get_project(project_name)
        folder = project.get_folder(folder_name)
        folder.set_items_approval_statuses(
            approval_status=ApprovalStatus[
                approval_status if approval_status else "None"
            ],
            items=items,
        )