Add custom field names input validation.

Author: VaghinakDev

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

@@ -36,7 +36,6 @@ class ProjectFilters(BaseFilters):
     status__ne: Literal["NotStarted", "InProgress", "Completed", "OnHold"]
     status__in: List[Literal["NotStarted", "InProgress", "Completed", "OnHold"]]
     status__notin: List[Literal["NotStarted", "InProgress", "Completed", "OnHold"]]
-    custom_field: Optional[str]  # dummy field to pass first level validation
 
 
 class UserFilters(TypedDict, total=False):
@@ -51,4 +50,3 @@ class UserFilters(TypedDict, total=False):
     state__in: Optional[List[Literal["CONFIRMED", "PENDING"]]]
     role: Optional[Literal["admin", "contributor"]]
     role__in: Optional[List[Literal["admin", "contributor"]]]
-    custom_field: Optional[str]  # dummy field to pass first level validation

src/superannotate/lib/core/jsx_conditions.py

@@ -14,6 +14,7 @@ class OperatorEnum(str, Enum):
     NE = "$ne"
     GT = "$gt"
     LT = "$lt"
+    IS = "$is"
     CONTAINS = "$cont"
     STARTS = "$starts"
     ENDS = "$ends"

src/superannotate/lib/infrastructure/controller.py

@@ -48,6 +48,7 @@
 from lib.core.response import Response
 from lib.core.service_types import PROJECT_TYPE_RESPONSE_MAP
 from lib.core.usecases import serialize_item_entity
+from lib.infrastructure.custom_entities import generate_schema
 from lib.infrastructure.helpers import timed_lru_cache
 from lib.infrastructure.query_builder import FieldValidationHandler
 from lib.infrastructure.query_builder import IncludeHandler
@@ -360,9 +361,15 @@ def list_projects(
         include: List[str] = None,
         **filters: Unpack[ProjectFilters],
     ) -> List[ProjectEntity]:
+        valid_fields = generate_schema(
+            ProjectFilters.__annotations__,
+            self.service_provider.get_custom_fields_templates(
+                CustomFieldEntityEnum.PROJECT
+            ),
+        )
         chain = QueryBuilderChain(
             [
-                FieldValidationHandler(ProjectFilters.__annotations__.keys()),
+                FieldValidationHandler(valid_fields.keys()),
                 ProjectFilterHandler(
                     self.service_provider, entity=CustomFieldEntityEnum.PROJECT
                 ),

src/superannotate/lib/infrastructure/custom_entities.py

@@ -0,0 +1,47 @@
+from lib.core.enums import CustomFieldType
+from lib.core.jsx_conditions import OperatorEnum
+from typing_extensions import Any
+
+
+FIELD_TYPE_SUPPORTED_OPERATIONS_MAPPING = {
+    CustomFieldType.Text: [
+        OperatorEnum.EQ,
+        OperatorEnum.IN,
+        OperatorEnum.NOTIN,
+        OperatorEnum.CONTAINS,
+    ],
+    CustomFieldType.MULTI_SELECT: [OperatorEnum.EQ, OperatorEnum.NOTIN],
+    CustomFieldType.SINGLE_SELECT: [
+        OperatorEnum.EQ,
+        OperatorEnum.IN,
+        OperatorEnum.NOTIN,
+        OperatorEnum.CONTAINS,
+    ],
+    CustomFieldType.DATE_PICKER: [OperatorEnum.GT, OperatorEnum.LT, OperatorEnum.EQ],
+    CustomFieldType.NUMERIC: [
+        OperatorEnum.GT,
+        OperatorEnum.LT,
+        OperatorEnum.EQ,
+        OperatorEnum.IN,
+        OperatorEnum.NOTIN,
+    ],
+}
+
+# todo implement
+FIELD_TYPE_BED_OPERATIONS_MAPPING = {
+    CustomFieldType.MULTI_SELECT: {OperatorEnum.EQ: OperatorEnum.IN},
+    CustomFieldType.DATE_PICKER: {OperatorEnum.EQ: OperatorEnum.IS},
+}
+
+
+def generate_schema(base_schema: dict, custom_field_templates) -> dict:
+    annotations = base_schema
+    for custom_field_template in custom_field_templates:
+        for operator in FIELD_TYPE_SUPPORTED_OPERATIONS_MAPPING[
+            CustomFieldType(custom_field_template["component_id"])
+        ]:
+            condition = f"custom_field__{custom_field_template['name']}"
+            if operator != OperatorEnum.EQ:
+                condition = condition + f"__{operator.name.lower()}"
+            annotations[condition] = Any
+    return annotations

src/superannotate/lib/infrastructure/query_builder.py

@@ -75,10 +75,7 @@ def __init__(self, valid_fields: Iterable[str]):
 
     def handle(self, filters: Dict[str, Any], query: Query = None) -> Query:
         for param in filters.keys():
-            if (
-                not param.startswith("custom_field__")
-                and param not in self._valid_fields
-            ):
+            if param not in self._valid_fields:
                 raise AppException("Invalid filter param provided.")
         return super().handle(filters, query)
 

src/superannotate/lib/infrastructure/serviceprovider.py

@@ -69,6 +69,11 @@ def __init__(self, client: HttpClient):
             5, self.work_management
         )
 
+    def get_custom_fields_templates(self, entity: CustomFieldEntityEnum):
+        return self._cached_work_management_repository.list_templates(
+            self.client.team_id, entity=entity
+        )
+
     def list_custom_field_names(self, entity: CustomFieldEntityEnum) -> List[str]:
         return self._cached_work_management_repository.list_custom_field_names(
             self.client.team_id, entity=entity

src/superannotate/lib/infrastructure/utils.py

@@ -138,6 +138,7 @@ def sync(self, team_id):
             "custom_fields_name_id_map": custom_fields_name_id_map,
             "custom_fields_id_name_map": custom_fields_id_name_map,
             "custom_fields_id_component_id_map": custom_fields_id_component_id_map,
+            "templates": response.data["data"],
         }
         self._update_cache_timestamp(team_id)
 
@@ -228,3 +229,10 @@ def list_custom_field_names(
         else:
             custom_field_data = self._user_custom_field_cache.get(team_id)
         return list(custom_field_data["custom_fields_name_id_map"].keys())
+
+    def list_templates(self, team_id: int, entity: CustomFieldEntityEnum):
+        if entity == CustomFieldEntityEnum.PROJECT:
+            custom_field_data = self._project_custom_field_cache.get(team_id)
+        else:
+            custom_field_data = self._user_custom_field_cache.get(team_id)
+        return custom_field_data["templates"]