feat(LAB-4081): add "is not" operator filters to Python SDK for status, step and step status

Author: paulruelle

src/kili/adapters/kili_api_gateway/asset/mappers.py

@@ -28,6 +28,7 @@ def asset_where_mapper(filters: AssetFilters):
         "externalIdStrictlyIn": filters.external_id_strictly_in,
         "externalIdIn": filters.external_id_in,
         "statusIn": filters.status_in,
+        "statusNotIn": filters.status_not_in,
         "consensusMarkGte": filters.consensus_mark_gte,
         "consensusMarkLte": filters.consensus_mark_lte,
         "honeypotMarkGte": filters.honeypot_mark_gte,
@@ -65,5 +66,7 @@ def asset_where_mapper(filters: AssetFilters):
             "status": filters.issue_status,
         },
         "stepIdIn": filters.step_id_in,
+        "stepIdNotIn": filters.step_id_not_in,
         "stepStatusIn": filters.step_status_in,
+        "stepStatusNotIn": filters.step_status_not_in,
     }

src/kili/domain/asset/asset.py

@@ -62,17 +62,23 @@ class AssetFilters:
     issue_status: Optional["IssueStatus"] = None
     skipped: Optional[bool] = None
     status_in: Optional[ListOrTuple[AssetStatus]] = None
+    status_not_in: Optional[ListOrTuple[AssetStatus]] = None
     step_id_in: Optional[ListOrTuple[str]] = None
+    step_id_not_in: Optional[ListOrTuple[str]] = None
     step_status_in: Optional[ListOrTuple[StatusInStep]] = None
+    step_status_not_in: Optional[ListOrTuple[StatusInStep]] = None
 
 
 class AssetWorkflowFilters(TypedDict):
     """Asset filters relative to worklow."""
 
     skipped: Optional[bool]
     status_in: Optional[ListOrTuple[AssetStatus]]
+    status_not_in: Optional[ListOrTuple[AssetStatus]]
     step_name_in: Optional[ListOrTuple[str]]
+    step_name_not_in: Optional[ListOrTuple[str]]
     step_status_in: Optional[ListOrTuple[StatusInStep]]
+    step_status_not_in: Optional[ListOrTuple[StatusInStep]]
 
 
 def get_asset_default_fields(

src/kili/domain/asset/helpers.py

@@ -34,24 +34,40 @@ def check_asset_workflow_arguments(
 ) -> None:
     """Check asset workflow parameters relative to the project workflow version."""
     step_name_in = asset_workflow_filters.get("step_name_in")
+    step_name_not_in = asset_workflow_filters.get("step_name_not_in")
     step_status_in = asset_workflow_filters.get("step_status_in")
+    step_status_not_in = asset_workflow_filters.get("step_status_not_in")
     status_in = asset_workflow_filters.get("status_in")
+    status_not_in = asset_workflow_filters.get("status_not_in")
     skipped = asset_workflow_filters.get("skipped")
 
     if project_workflow_version == "V2":
         if step_status_in is not None and status_in is not None:
             raise ValueError(
                 "Filters step_status_in and status_in both given : only use filter step_status_in for this project."
             )
+        if step_status_not_in is not None and status_not_in is not None:
+            raise ValueError(
+                "Filters step_status_not_in and status_not_in both given : only use filter step_status_not_in for this project."
+            )
         if step_name_in is not None and status_in is not None:
             raise ValueError(
                 "Filters step_name_in and status_in both given : use filter step_status_in instead of status_in for this project."  # pylint: disable=line-too-long
             )
+        if step_name_not_in is not None and status_not_in is not None:
+            raise ValueError(
+                "Filters step_name_not_in and status_not_in both given : use filter step_status_not_in instead of status_not_in for this project."  # pylint: disable=line-too-long
+            )
         if status_in is not None:
             warnings.warn(
                 "Filter status_in given : use filters step_status_in and step_name_in instead for this project.",
                 stacklevel=1,
             )
+        if status_not_in is not None:
+            warnings.warn(
+                "Filter status_not_in given : use filters step_status_not_in and step_name_not_in instead for this project.",
+                stacklevel=1,
+            )
         if skipped is not None:
             warnings.warn(
                 "Filter skipped given : only use filter step_status_in with the SKIPPED step status instead for this project",  # pylint: disable=line-too-long
@@ -64,3 +80,7 @@ def check_asset_workflow_arguments(
         raise ValueError(
             "Filters step_name_in and/or step_status_in given : use filter status_in for this project."
         )
+    if step_name_not_in is not None or step_status_not_in is not None:
+        raise ValueError(
+            "Filters step_name_not_in and/or step_status_not_in given : use filter status_not_in for this project."
+        )

src/kili/domain_api/assets.py

@@ -88,8 +88,11 @@ class AssetFilter(TypedDict, total=False):
     metadata_where: Optional[Dict[str, Any]]
     skipped: Optional[bool]
     status_in: Optional[List[AssetStatus]]
+    status_not_in: Optional[List[AssetStatus]]
     step_name_in: Optional[List[str]]
+    step_name_not_in: Optional[List[str]]
     step_status_in: Optional[List[StatusInStep]]
+    step_status_not_in: Optional[List[StatusInStep]]
     updated_at_gte: Optional[str]
     updated_at_lte: Optional[str]
 

src/kili/presentation/client/asset.py

@@ -110,8 +110,11 @@ def assets(
         label_output_format: Literal["dict", "parsed_label"] = "dict",
         skipped: Optional[bool] = None,
         status_in: Optional[List[AssetStatus]] = None,
+        status_not_in: Optional[List[AssetStatus]] = None,
         step_name_in: Optional[List[str]] = None,
+        step_name_not_in: Optional[List[str]] = None,
         step_status_in: Optional[List[StatusInStep]] = None,
+        step_status_not_in: Optional[List[StatusInStep]] = None,
         *,
         as_generator: Literal[True],
     ) -> Generator[Dict, None, None]:
@@ -176,8 +179,11 @@ def assets(
         label_output_format: Literal["dict", "parsed_label"] = "dict",
         skipped: Optional[bool] = None,
         status_in: Optional[List[AssetStatus]] = None,
+        status_not_in: Optional[List[AssetStatus]] = None,
         step_name_in: Optional[List[str]] = None,
+        step_name_not_in: Optional[List[str]] = None,
         step_status_in: Optional[List[StatusInStep]] = None,
+        step_status_not_in: Optional[List[StatusInStep]] = None,
         *,
         as_generator: Literal[False] = False,
     ) -> List[Dict]:
@@ -242,8 +248,11 @@ def assets(
         label_output_format: Literal["dict", "parsed_label"] = "dict",
         skipped: Optional[bool] = None,
         status_in: Optional[List[AssetStatus]] = None,
+        status_not_in: Optional[List[AssetStatus]] = None,
         step_name_in: Optional[List[str]] = None,
+        step_name_not_in: Optional[List[str]] = None,
         step_status_in: Optional[List[StatusInStep]] = None,
+        step_status_not_in: Optional[List[StatusInStep]] = None,
         *,
         as_generator: bool = False,
     ) -> Union[Iterable[Dict], "pd.DataFrame"]:
@@ -313,10 +322,18 @@ def assets(
             status_in: Returned assets should have a status that belongs to that list, if given.
                 Possible choices: `TODO`, `ONGOING`, `LABELED`, `TO_REVIEW` or `REVIEWED`.
                 Only applicable if the project is in the WorkflowV1 (legacy).
+            status_not_in: Returned assets should have a status that does not belong to that list, if given.
+                Possible choices: `TODO`, `ONGOING`, `LABELED`, `TO_REVIEW` or `REVIEWED`.
+                Only applicable if the project is in the WorkflowV1 (legacy).
             step_name_in: Returned assets are in the step whose name belong to that list, if given.
                 Only applicable if the project is in WorkflowV2.
+            step_name_not_in: Returned assets are in the step whose name does not belong to that list, if given.
+                Only applicable if the project is in WorkflowV2.
             step_status_in: Returned assets have the status in their step that belongs to that list, if given.
                 Only applicable if the project is in WorkflowV2.
+            step_status_not_in: Returned assets have the status in their step that does not belong to that list, if given.
+                Possible choices: `TO_DO`, `DOING`, `PARTIALLY_DONE`, `REDO`, `DONE`, `SKIPPED`.
+                Only applicable if the project is in WorkflowV2.
 
         !!! info "Dates format"
             Date strings should have format: "YYYY-MM-DD"
@@ -432,26 +449,38 @@ def assets(
             )
 
         step_id_in = None
+        step_id_not_in = None
         if (
             step_name_in is not None
+            or step_name_not_in is not None
             or step_status_in is not None
+            or step_status_not_in is not None
             or status_in is not None
+            or status_not_in is not None
             or skipped is not None
         ):
             check_asset_workflow_arguments(
                 project_workflow_version=project_workflow_version,
                 asset_workflow_filters={
                     "skipped": skipped,
                     "status_in": status_in,
+                    "status_not_in": status_not_in,
                     "step_name_in": step_name_in,
+                    "step_name_not_in": step_name_not_in,
                     "step_status_in": step_status_in,
+                    "step_status_not_in": step_status_not_in,
                 },
             )
             if project_workflow_version == "V2" and step_name_in is not None:
                 step_id_in = extract_step_ids_from_project_steps(
                     project_steps=project_steps,
                     step_name_in=step_name_in,
                 )
+            if project_workflow_version == "V2" and step_name_not_in is not None:
+                step_id_not_in = extract_step_ids_from_project_steps(
+                    project_steps=project_steps,
+                    step_name_in=step_name_not_in,
+                )
 
         asset_use_cases = AssetUseCases(self.kili_api_gateway)
         filters = AssetFilters(
@@ -497,7 +526,10 @@ def assets(
             issue_status=issue_status,
             issue_type=issue_type,
             step_id_in=step_id_in,
+            step_id_not_in=step_id_not_in,
             step_status_in=step_status_in,
+            step_status_not_in=step_status_not_in,
+            status_not_in=status_not_in,
         )
         assets_gen = asset_use_cases.list_assets(
             filters,
@@ -570,7 +602,10 @@ def count_assets(
         external_id_strictly_in: Optional[List[str]] = None,
         external_id_in: Optional[List[str]] = None,
         step_name_in: Optional[List[str]] = None,
+        step_name_not_in: Optional[List[str]] = None,
         step_status_in: Optional[List[StatusInStep]] = None,
+        step_status_not_in: Optional[List[StatusInStep]] = None,
+        status_not_in: Optional[List[AssetStatus]] = None,
     ) -> int:
         # pylint: disable=line-too-long
         """Count and return the number of assets with the given constraints.
@@ -629,9 +664,17 @@ def count_assets(
                 For example, with `external_id_in=['abc']`, any asset with an external id containing `'abc'` will be returned.
             step_name_in: Returned assets are in a step whose name belong to that list, if given.
                 Only applicable if the project is in WorkflowV2.
+            step_name_not_in: Returned assets are in a step whose name does not belong to that list, if given.
+                Only applicable if the project is in WorkflowV2.
             step_status_in: Returned assets have the status of their step that belongs to that list, if given.
                 Possible choices: `TO_DO`, `DOING`, `PARTIALLY_DONE`, `REDO`, `DONE`, `SKIPPED`.
                 Only applicable if the project is in WorkflowV2.
+            step_status_not_in: Returned assets have the status of their step that does not belong to that list, if given.
+                Possible choices: `TO_DO`, `DOING`, `PARTIALLY_DONE`, `REDO`, `DONE`, `SKIPPED`.
+                Only applicable if the project is in WorkflowV2.
+            status_not_in: Returned assets should have a status that does not belong to that list, if given.
+                Possible choices: `TODO`, `ONGOING`, `LABELED`, `TO_REVIEW` or `REVIEWED`.
+                Only applicable if the project is in WorkflowV1 (legacy).
 
         !!! info "Dates format"
             Date strings should have format: "YYYY-MM-DD"
@@ -695,7 +738,15 @@ def count_assets(
                 )
 
         step_id_in = None
-        if status_in is not None or step_name_in is not None or step_status_in is not None:
+        step_id_not_in = None
+        if (
+            status_in is not None
+            or status_not_in is not None
+            or step_name_in is not None
+            or step_name_not_in is not None
+            or step_status_in is not None
+            or step_status_not_in is not None
+        ):
             project_use_cases = ProjectUseCases(self.kili_api_gateway)
             (
                 project_steps,
@@ -706,8 +757,11 @@ def count_assets(
                 asset_workflow_filters={
                     "skipped": skipped,
                     "step_name_in": step_name_in,
+                    "step_name_not_in": step_name_not_in,
                     "step_status_in": step_status_in,
+                    "step_status_not_in": step_status_not_in,
                     "status_in": status_in,
+                    "status_not_in": status_not_in,
                 },
             )
 
@@ -716,6 +770,11 @@ def count_assets(
                     project_steps=project_steps,
                     step_name_in=step_name_in,
                 )
+            if project_workflow_version == "V2" and step_name_not_in is not None:
+                step_id_not_in = extract_step_ids_from_project_steps(
+                    project_steps=project_steps,
+                    step_name_in=step_name_not_in,
+                )
 
         filters = AssetFilters(
             project_id=ProjectId(project_id),
@@ -760,7 +819,10 @@ def count_assets(
             issue_status=issue_status,
             issue_type=issue_type,
             step_id_in=step_id_in,
+            step_id_not_in=step_id_not_in,
             step_status_in=step_status_in,
+            step_status_not_in=step_status_not_in,
+            status_not_in=status_not_in,
         )
         asset_use_cases = AssetUseCases(self.kili_api_gateway)
         return asset_use_cases.count_assets(filters)