Added integrations

Author: VaghinakDev

src/superannotate/__init__.py

@@ -21,6 +21,8 @@
 from superannotate.lib.app.interface.sdk_interface import aggregate_annotations_as_df
 from superannotate.lib.app.interface.sdk_interface import assign_folder
 from superannotate.lib.app.interface.sdk_interface import assign_images
+from superannotate.lib.app.interface.sdk_interface import get_integrations
+from superannotate.lib.app.interface.sdk_interface import attach_items_from_integrated_storage
 from superannotate.lib.app.interface.sdk_interface import (
     attach_document_urls_to_project,
 )
@@ -135,6 +137,9 @@
     # annotations
     "get_annotations",
     "get_annotations_per_frame",
+    # integrations
+    "get_integrations",
+    "attach_items_from_integrated_storage",
     # converters
     "convert_json_version",
     "import_annotation",

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

@@ -34,6 +34,7 @@
 from lib.app.serializers import SettingsSerializer
 from lib.app.serializers import TeamSerializer
 from lib.core import LIMITED_FUNCTIONS
+from lib.core.entities.integrations import IntegrationEntity
 from lib.core.entities.project_entities import AnnotationClassEntity
 from lib.core.enums import ImageQuality
 from lib.core.exceptions import AppException
@@ -2923,3 +2924,49 @@ def upload_priority_scores(project: NotEmptyStr, scores: List[PriorityScore]):
     if response.errors:
         raise AppException(response.errors)
     return response.data
+
+
+@Trackable
+@validate_arguments
+def get_integrations():
+    """Get all integrations per team
+
+    :return: metadata objects of all integrations of the team.
+    :rtype: list of dicts
+    """
+    response = Controller.get_default().get_integrations()
+    if response.errors:
+        raise AppException(response.errors)
+    integrations = response.data
+    return BaseSerializers.serialize_iterable(integrations, ("name", "type", "root"))
+
+
+@Trackable
+@validate_arguments
+def attach_items_from_integrated_storage(
+        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 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
+
+    :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_name, folder_name = extract_project_folder(project)
+    if isinstance(integration, str):
+        integration = IntegrationEntity(name=integration)
+    response = Controller.get_default().attach_integrations(project_name, folder_name, integration, folder_path)
+    if response.errors:
+        raise AppException(response.errors)

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

@@ -1,6 +1,7 @@
 import lib.core as constances
 from lib.app.helpers import extract_project_folder
 from lib.core.enums import ProjectType
+from lib.core.entities import IntegrationEntity
 from lib.infrastructure.controller import Controller
 
 
@@ -1238,3 +1239,36 @@ def upload_priority_scores(*args, **kwargs):
         "event_name": "upload_priority_scores",
         "properties": {"Score Count": len(scores)},
     }
+
+
+def get_integrations(*args, **kwargs):
+    return {
+        "event_name": "get_integrations",
+        "properties": {},
+    }
+
+
+def attach_items_from_integrated_storage(*args, **kwargs):
+    project = kwargs.get("project")
+    integration = kwargs.get("integration")
+    folder_path = kwargs.get("folder_path")
+    if not project:
+        project = args[0]
+    if not integration:
+        integration = args[1]
+    if not folder_path:
+        folder_path = args[2]
+
+    project_name, _ = extract_project_folder(project)
+    if isinstance(integration, str):
+        integration = IntegrationEntity(name=integration)
+    project = Controller.get_default().get_project_metadata(project_name)
+    return {
+        "event_name": "attach_items_from_integrated_storage",
+        "properties": {
+            "project_type": ProjectType.get_name(project.project_type),
+            "integration_name": integration.name,
+            "folder_path": bool(folder_path)
+        },
+    }
+

src/superannotate/lib/app/serializers.py

@@ -1,4 +1,8 @@
 from abc import ABC
+from typing import List
+from typing import Set
+from typing import Union
+from typing import Any
 
 import superannotate.lib.core as constance
 from pydantic import BaseModel
@@ -11,12 +15,37 @@ class BaseSerializers(ABC):
     def __init__(self, entity: BaseEntity):
         self._entity = entity
 
-    def serialize(self):
-        if isinstance(self._entity, dict):
-            return self._entity
-        if isinstance(self._entity, BaseModel):
-            return self._entity.dict(by_alias=True)
-        return self._entity.to_dict()
+    def serialize(self, fields: List[str] = None, by_alias: bool = True, flat: bool = False):
+            return self._serialize(self._entity, fields, by_alias, flat)
+
+    @staticmethod
+    def _serialize(entity: Any, fields: List[str] = None, by_alias: bool = False, flat: bool = False):
+        if isinstance(entity, dict):
+            return entity
+        if isinstance(entity, BaseModel):
+            if fields:
+                fields = set(fields)
+                if len(fields) == 1:
+                    if flat:
+                        return entity.dict(include=fields, by_alias=by_alias)[next(iter(fields))]
+                    else:
+                        return entity.dict(include=fields, by_alias=by_alias)
+                return entity.dict(include=fields, by_alias=by_alias)
+            return entity.dict(by_alias=by_alias)
+        return entity.to_dict()
+
+    @classmethod
+    def serialize_iterable(
+            cls,
+            data: List[Any],
+            fields: Union[List[str], Set[str]] = None,
+            by_alias: bool = False,
+            flat: bool = False
+    ) -> List[Any]:
+        serialized_data = []
+        for i in data:
+            serialized_data.append(cls._serialize(i, fields, by_alias, flat))
+        return serialized_data
 
 
 class UserSerializer(BaseSerializers):

src/superannotate/lib/core/entities/__init__.py

@@ -11,6 +11,7 @@
 from lib.core.entities.project_entities import TeamEntity
 from lib.core.entities.project_entities import UserEntity
 from lib.core.entities.project_entities import WorkflowEntity
+from lib.core.entities.integrations import IntegrationEntity
 from superannotate_schemas.schemas.internal.document import DocumentAnnotation
 from superannotate_schemas.schemas.internal.pixel import PixelAnnotation
 from superannotate_schemas.schemas.internal.vector import VectorAnnotation
@@ -33,6 +34,7 @@
     "UserEntity",
     "TeamEntity",
     "MLModelEntity",
+    "IntegrationEntity",
     # annotations
     "DocumentAnnotation",
     "VideoAnnotation",

src/superannotate/lib/core/entities/base.py

@@ -0,0 +1,10 @@
+from datetime import datetime
+
+from pydantic import BaseModel
+from pydantic import Field
+
+
+class TimedBaseModel(BaseModel):
+    created_at: datetime = Field(None, alias="createdAt")
+    updated_at: datetime = Field(None, alias="updatedAt")
+

src/superannotate/lib/core/entities/integrations.py

@@ -0,0 +1,14 @@
+from pydantic import Field
+from lib.core.entities.base import TimedBaseModel
+
+
+class IntegrationEntity(TimedBaseModel):
+    id: int = None
+    user_id: str = None
+    name: str
+    type: str = "aws"
+    root: str = Field(None, alias="bucket_name")
+    source: int = None
+
+    class Config:
+        arbitrary_types_allowed = True