Added create/get/delete cusotm fields

Author: VaghinakDev

docs/source/superannotate.sdk.rst

@@ -79,6 +79,15 @@ ______
 
 ----------
 
+Custom Metadata
+______
+
+.. automethod:: superannotate.SAClient.create_custom_fields
+.. automethod:: superannotate.SAClient.get_custom_fields
+.. automethod:: superannotate.SAClient.delete_custom_fields
+
+----------
+
 Subsets
 ______
 

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

@@ -3,7 +3,6 @@
 import json
 import os
 import tempfile
-import warnings
 from pathlib import Path
 from typing import Callable
 from typing import Iterable
@@ -2634,3 +2633,177 @@ def get_subsets(self, project: Union[NotEmptyStr, dict]):
         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” or “date”
+        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)
+        response = self.controller.create_custom_schema(
+            project_name=project, schema=fields
+        )
+        if response.errors:
+            raise AppException(response.errors)
+        return response.data
+
+    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)
+        response = self.controller.get_custom_schema(project_name=project)
+        if response.errors:
+            raise AppException(response.errors)
+        return response.data
+
+    def delete_custom_fields(self, project: NotEmptyStr, fields: list):
+        """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)
+        response = self.controller.delete_custom_schema(
+            project_name=project, fields=fields
+        )
+        if response.errors:
+            raise AppException(response.errors)
+        return response.data

src/superannotate/lib/core/serviceproviders.py

@@ -357,3 +357,16 @@ def validate_saqul_query(self, team_id: int, project_id: int, query: str) -> dic
 
     def list_sub_sets(self, team_id: int, project_id: int) -> ServiceResponse:
         raise NotImplementedError
+
+    def create_custom_schema(
+        self, team_id: int, project_id: int, schema: dict
+    ) -> ServiceResponse:
+        raise NotImplementedError
+
+    def get_custom_schema(self, team_id: int, project_id: int) -> ServiceResponse:
+        raise NotImplementedError
+
+    def delete_custom_schema(
+        self, team_id: int, project_id: int, fields: List[str]
+    ) -> ServiceResponse:
+        raise NotImplementedError

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

@@ -1,4 +1,5 @@
 from lib.core.usecases.annotations import *  # noqa: F403 F401
+from lib.core.usecases.custom_fields import *  # noqa: F403 F401
 from lib.core.usecases.folders import *  # noqa: F403 F401
 from lib.core.usecases.images import *  # noqa: F403 F401
 from lib.core.usecases.integrations import *  # noqa: F403 F401

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

@@ -0,0 +1,82 @@
+from typing import List
+
+from lib.core.entities import ProjectEntity
+from lib.core.reporter import Reporter
+from lib.core.response import Response
+from lib.core.serviceproviders import SuperannotateServiceProvider
+from lib.core.usecases import BaseReportableUseCase
+
+
+class CreateCustomSchemaUseCase(BaseReportableUseCase):
+    def __init__(
+        self,
+        reporter: Reporter,
+        project: ProjectEntity,
+        schema: dict,
+        backend_client: SuperannotateServiceProvider,
+    ):
+        super().__init__(reporter)
+        self._project = project
+        self._schema = schema
+        self._backend_client = backend_client
+
+    def execute(self) -> Response:
+        response = self._backend_client.create_custom_schema(
+            team_id=self._project.team_id,
+            project_id=self._project.id,
+            schema=self._schema,
+        )
+        if response.ok:
+            self._response.data = response.data
+        else:
+            self._response.errors = response.error
+        return self._response
+
+
+class GetCustomSchemaUseCase(BaseReportableUseCase):
+    def __init__(
+        self,
+        reporter: Reporter,
+        project: ProjectEntity,
+        backend_client: SuperannotateServiceProvider,
+    ):
+        super().__init__(reporter)
+        self._project = project
+        self._backend_client = backend_client
+
+    def execute(self) -> Response:
+        response = self._backend_client.get_custom_schema(
+            team_id=self._project.team_id,
+            project_id=self._project.id,
+        )
+        if response.ok:
+            self._response.data = response.data
+        else:
+            self._response.errors = response.error
+        return self._response
+
+
+class DeleteCustomSchemaUseCase(BaseReportableUseCase):
+    def __init__(
+        self,
+        reporter: Reporter,
+        project: ProjectEntity,
+        fields: List[str],
+        backend_client: SuperannotateServiceProvider,
+    ):
+        super().__init__(reporter)
+        self._project = project
+        self._fields = fields
+        self._backend_client = backend_client
+
+    def execute(self) -> Response:
+        response = self._backend_client.delete_custom_schema(
+            team_id=self._project.team_id,
+            project_id=self._project.id,
+            fields=self._fields,
+        )
+        if response.ok:
+            self._response.data = response.data
+        else:
+            self._response.errors = response.error
+        return self._response

src/superannotate/lib/infrastructure/controller.py

@@ -1658,3 +1658,33 @@ def list_subsets(self, project_name: str):
             backend_client=self.backend_client,
         )
         return use_case.execute()
+
+    def create_custom_schema(self, project_name: str, schema: dict):
+        project = self._get_project(project_name)
+
+        use_case = usecases.CreateCustomSchemaUseCase(
+            reporter=self.get_default_reporter(),
+            project=project,
+            schema=schema,
+            backend_client=self.backend_client,
+        )
+        return use_case.execute()
+
+    def get_custom_schema(self, project_name: str):
+        project = self._get_project(project_name)
+        use_case = usecases.GetCustomSchemaUseCase(
+            reporter=self.get_default_reporter(),
+            project=project,
+            backend_client=self.backend_client,
+        )
+        return use_case.execute()
+
+    def delete_custom_schema(self, project_name: str, fields: List[str]):
+        project = self._get_project(project_name)
+        use_case = usecases.DeleteCustomSchemaUseCase(
+            reporter=self.get_default_reporter(),
+            project=project,
+            fields=fields,
+            backend_client=self.backend_client,
+        )
+        return use_case.execute()

src/superannotate/lib/infrastructure/services.py

@@ -244,6 +244,8 @@ class SuperannotateBackendService(BaseBackendService):
     URL_SAQUL_QUERY = "/images/search/advanced"
     URL_VALIDATE_SAQUL_QUERY = "/images/parse/query/advanced"
     URL_LIST_SUBSETS = "/project/{project_id}/subset"
+    URL_CREATE_CUSTOM_SCHEMA = "/project/{project_id}/custom/metadata/schema"
+    URL_GET_CUSTOM_SCHEMA = "/project/{project_id}/custom/metadata/schema"
 
     def upload_priority_scores(
         self, team_id: int, project_id: int, folder_id: int, priorities: list
@@ -1191,3 +1193,42 @@ def list_sub_sets(self, team_id: int, project_id: int) -> ServiceResponse:
             params=dict(team_id=team_id),
             content_type=List[entities.SubSetEntity],
         )
+
+    def create_custom_schema(
+        self, team_id: int, project_id: int, schema: dict
+    ) -> ServiceResponse:
+        return self._request(
+            urljoin(
+                self.api_url,
+                self.URL_CREATE_CUSTOM_SCHEMA.format(project_id=project_id),
+            ),
+            "post",
+            params=dict(team_id=team_id),
+            data=dict(data=schema),
+            content_type=ServiceResponse,
+        )
+
+    def get_custom_schema(self, team_id: int, project_id: int) -> ServiceResponse:
+        return self._request(
+            urljoin(
+                self.api_url,
+                self.URL_CREATE_CUSTOM_SCHEMA.format(project_id=project_id),
+            ),
+            "get",
+            params=dict(team_id=team_id),
+            content_type=ServiceResponse,
+        )
+
+    def delete_custom_schema(
+        self, team_id: int, project_id: int, fields: List[str]
+    ) -> ServiceResponse:
+        return self._request(
+            urljoin(
+                self.api_url,
+                self.URL_CREATE_CUSTOM_SCHEMA.format(project_id=project_id),
+            ),
+            "delete",
+            params=dict(team_id=team_id),
+            data=dict(custom_fields=fields),
+            content_type=ServiceResponse,
+        )