feat(LAB-4339): add step_name_and_status_in / not_in filters

Author: Josselin BUILS

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

@@ -68,4 +68,20 @@ def asset_where_mapper(filters: AssetFilters):
         "stepIdNotIn": filters.step_id_not_in,
         "stepStatusIn": filters.step_status_in,
         "stepStatusNotIn": filters.step_status_not_in,
+        "stepIdAndStatusIn": (
+            [
+                {"stepId": step_id, "status": step_status}
+                for step_id, step_status in filters.step_id_and_status_in
+            ]
+            if filters.step_id_and_status_in is not None
+            else None
+        ),
+        "stepIdAndStatusNotIn": (
+            [
+                {"stepId": step_id, "status": step_status}
+                for step_id, step_status in filters.step_id_and_status_not_in
+            ]
+            if filters.step_id_and_status_not_in is not None
+            else None
+        ),
     }

src/kili/domain/asset/asset.py

@@ -62,6 +62,8 @@ class AssetFilters:
     issue_status: Optional["IssueStatus"] = None
     skipped: Optional[bool] = None
     status_in: Optional[ListOrTuple[AssetStatus]] = None
+    step_id_and_status_in: Optional[list[tuple[str, StatusInStep]]] = None
+    step_id_and_status_not_in: Optional[list[tuple[str, StatusInStep]]] = None
     step_id_in: Optional[ListOrTuple[str]] = None
     step_id_not_in: Optional[ListOrTuple[str]] = None
     step_status_in: Optional[ListOrTuple[StatusInStep]] = None
@@ -73,6 +75,8 @@ class AssetWorkflowFilters(TypedDict, total=False):
 
     skipped: Optional[bool]
     status_in: Optional[ListOrTuple[AssetStatus]]
+    step_name_and_status_in: Optional[list[tuple[str, StatusInStep]]]
+    step_name_and_status_not_in: Optional[list[tuple[str, StatusInStep]]]
     step_name_in: Optional[ListOrTuple[str]]
     step_name_not_in: Optional[ListOrTuple[str]]
     step_status_in: Optional[ListOrTuple[StatusInStep]]

src/kili/domain/asset/helpers.py

@@ -33,6 +33,8 @@ def check_asset_workflow_arguments(
     project_workflow_version: WorkflowVersion, asset_workflow_filters: AssetWorkflowFilters
 ) -> None:
     """Check asset workflow parameters relative to the project workflow version."""
+    step_name_and_status_in = asset_workflow_filters.get("step_name_and_status_in")
+    step_name_and_status_not_in = asset_workflow_filters.get("step_name_and_status_not_in")
     step_name_in = asset_workflow_filters.get("step_name_in")
     step_status_in = asset_workflow_filters.get("step_status_in")
     status_in = asset_workflow_filters.get("status_in")
@@ -64,3 +66,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_and_status_in is not None or step_name_and_status_not_in is not None:
+        raise ValueError(
+            "Filters step_name_and_status_in and/or step_name_and_status_not_in given : use filter status_in for this project."
+        )

src/kili/domain_api/assets.py

@@ -87,6 +87,8 @@ class AssetFilter(TypedDict, total=False):
     metadata_where: Optional[dict[str, Any]]
     skipped: Optional[bool]
     status_in: Optional[list[AssetStatus]]
+    step_name_and_status_in: Optional[list[tuple[str, StatusInStep]]]
+    step_name_and_status_not_in: Optional[list[tuple[str, StatusInStep]]]
     step_name_in: Optional[list[str]]
     step_name_not_in: Optional[list[str]]
     step_status_in: Optional[list[StatusInStep]]

src/kili/presentation/client/asset.py

@@ -32,6 +32,7 @@
     resolve_disable_tqdm,
 )
 from kili.presentation.client.helpers.filter_conversion import (
+    extract_step_id_and_status_filters_from_project_steps,
     extract_step_ids_from_project_steps,
 )
 from kili.use_cases.asset import AssetUseCases
@@ -157,6 +158,8 @@ def assets(
         label_output_format: Literal["dict", "parsed_label"] = "dict",
         skipped: Optional[bool] = None,
         status_in: Optional[list[AssetStatus]] = None,
+        step_name_and_status_in: Optional[list[tuple[str, StatusInStep]]] = None,
+        step_name_and_status_not_in: Optional[list[tuple[str, StatusInStep]]] = None,
         step_name_in: Optional[list[str]] = None,
         step_name_not_in: Optional[list[str]] = None,
         step_status_in: Optional[list[StatusInStep]] = None,
@@ -225,6 +228,8 @@ def assets(
         label_output_format: Literal["dict", "parsed_label"] = "dict",
         skipped: Optional[bool] = None,
         status_in: Optional[list[AssetStatus]] = None,
+        step_name_and_status_in: Optional[list[tuple[str, StatusInStep]]] = None,
+        step_name_and_status_not_in: Optional[list[tuple[str, StatusInStep]]] = None,
         step_name_in: Optional[list[str]] = None,
         step_name_not_in: Optional[list[str]] = None,
         step_status_in: Optional[list[StatusInStep]] = None,
@@ -293,6 +298,8 @@ def assets(
         label_output_format: Literal["dict", "parsed_label"] = "dict",
         skipped: Optional[bool] = None,
         status_in: Optional[list[AssetStatus]] = None,
+        step_name_and_status_in: Optional[list[tuple[str, StatusInStep]]] = None,
+        step_name_and_status_not_in: Optional[list[tuple[str, StatusInStep]]] = None,
         step_name_in: Optional[list[str]] = None,
         step_name_not_in: Optional[list[str]] = None,
         step_status_in: Optional[list[StatusInStep]] = None,
@@ -366,6 +373,10 @@ 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).
+            step_name_and_status_in: Returned assets match at least one of the given (step_name, step_status) pairs.
+                Only applicable if the project is in WorkflowV2.
+            step_name_and_status_not_in: Returned assets do not match any of the given (step_name, step_status) pairs.
+                Only applicable if the project is in WorkflowV2.
             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.
@@ -468,16 +479,21 @@ def assets(
                 stacklevel=1,
             )
 
+        step_id_and_status_in: Optional[list[tuple[str, StatusInStep]]] = None
+        step_id_and_status_not_in: Optional[list[tuple[str, StatusInStep]]] = None
         step_id_in = None
         step_id_not_in = None
-        has_step_filters = step_name_in is not None or step_name_not_in is not None
-        has_status_filters = (
-            step_status_in is not None
+        has_step_or_status_filters = (
+            step_name_in is not None
+            or step_name_not_in is not None
+            or step_name_and_status_in is not None
+            or step_name_and_status_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 skipped is not None
         )
-        if has_step_filters or has_status_filters:
+        if has_step_or_status_filters:
             check_asset_workflow_arguments(
                 project_workflow_version=project_workflow_version,
                 asset_workflow_filters={
@@ -487,6 +503,8 @@ def assets(
                     "step_name_not_in": step_name_not_in,
                     "step_status_in": step_status_in,
                     "step_status_not_in": step_status_not_in,
+                    "step_name_and_status_in": step_name_and_status_in,
+                    "step_name_and_status_not_in": step_name_and_status_not_in,
                 },
             )
             if project_workflow_version == "V2" and step_name_in is not None:
@@ -499,6 +517,16 @@ def assets(
                     project_steps=project_steps,
                     step_name_in=step_name_not_in,
                 )
+            if project_workflow_version == "V2" and step_name_and_status_in is not None:
+                step_id_and_status_in = extract_step_id_and_status_filters_from_project_steps(
+                    project_steps=project_steps,
+                    step_name_and_status_filters=step_name_and_status_in,
+                )
+            if project_workflow_version == "V2" and step_name_and_status_not_in is not None:
+                step_id_and_status_not_in = extract_step_id_and_status_filters_from_project_steps(
+                    project_steps=project_steps,
+                    step_name_and_status_filters=step_name_and_status_not_in,
+                )
 
         # Resolve disable_tqdm: function parameter > client global setting > function default
         disable_tqdm = resolve_disable_tqdm(disable_tqdm, getattr(self, "disable_tqdm", None))
@@ -550,6 +578,8 @@ def assets(
             step_id_not_in=step_id_not_in,
             step_status_in=step_status_in,
             step_status_not_in=step_status_not_in,
+            step_id_and_status_in=step_id_and_status_in,
+            step_id_and_status_not_in=step_id_and_status_not_in,
         )
         assets_gen = asset_use_cases.list_assets(
             filters,
@@ -625,6 +655,8 @@ def count_assets(
         step_name_not_in: Optional[list[str]] = None,
         step_status_in: Optional[list[StatusInStep]] = None,
         step_status_not_in: Optional[list[StatusInStep]] = None,
+        step_name_and_status_in: Optional[list[tuple[str, StatusInStep]]] = None,
+        step_name_and_status_not_in: Optional[list[tuple[str, StatusInStep]]] = None,
     ) -> int:
         # pylint: disable=line-too-long
         """Count and return the number of assets with the given constraints.
@@ -691,6 +723,10 @@ def count_assets(
             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.
+            step_name_and_status_in: Returned assets match at least one of the given (step_name, step_status) pairs.
+                Only applicable if the project is in WorkflowV2.
+            step_name_and_status_not_in: Returned assets do not match any of the given (step_name, step_status) pairs.
+                Only applicable if the project is in WorkflowV2.
 
         !!! info "Dates format"
             Date strings should have format: "YYYY-MM-DD"
@@ -734,11 +770,18 @@ def count_assets(
 
         step_id_in = None
         step_id_not_in = None
-        has_step_filters = step_name_in is not None or step_name_not_in is not None
-        has_status_filters = (
-            status_in is not None or step_status_in is not None or step_status_not_in is not None
+        step_id_and_status_in: Optional[list[tuple[str, StatusInStep]]] = None
+        step_id_and_status_not_in: Optional[list[tuple[str, StatusInStep]]] = None
+        has_step_or_status_filters = (
+            step_name_in is not None
+            or step_name_not_in is not None
+            or step_name_and_status_in is not None
+            or step_name_and_status_not_in is not None
+            or status_in is not None
+            or step_status_in is not None
+            or step_status_not_in is not None
         )
-        if has_step_filters or has_status_filters:
+        if has_step_or_status_filters:
             project_use_cases = ProjectUseCases(self.kili_api_gateway)
             (
                 project_steps,
@@ -749,6 +792,8 @@ def count_assets(
                 asset_workflow_filters={
                     "skipped": skipped,
                     "step_name_in": step_name_in,
+                    "step_name_and_status_in": step_name_and_status_in,
+                    "step_name_and_status_not_in": step_name_and_status_not_in,
                     "step_name_not_in": step_name_not_in,
                     "step_status_in": step_status_in,
                     "step_status_not_in": step_status_not_in,
@@ -766,6 +811,16 @@ def count_assets(
                     project_steps=project_steps,
                     step_name_in=step_name_not_in,
                 )
+            if project_workflow_version == "V2" and step_name_and_status_in is not None:
+                step_id_and_status_in = extract_step_id_and_status_filters_from_project_steps(
+                    project_steps=project_steps,
+                    step_name_and_status_filters=step_name_and_status_in,
+                )
+            if project_workflow_version == "V2" and step_name_and_status_not_in is not None:
+                step_id_and_status_not_in = extract_step_id_and_status_filters_from_project_steps(
+                    project_steps=project_steps,
+                    step_name_and_status_filters=step_name_and_status_not_in,
+                )
 
         filters = AssetFilters(
             project_id=ProjectId(project_id),
@@ -809,6 +864,8 @@ def count_assets(
             created_at_lte=created_at_lte,
             issue_status=issue_status,
             issue_type=issue_type,
+            step_id_and_status_in=step_id_and_status_in,
+            step_id_and_status_not_in=step_id_and_status_not_in,
             step_id_in=step_id_in,
             step_id_not_in=step_id_not_in,
             step_status_in=step_status_in,

src/kili/presentation/client/helpers/filter_conversion.py

@@ -1,6 +1,6 @@
 """Module for common argument validators across client methods."""
 
-
+from kili.domain.asset.asset import StatusInStep
 from kili.domain.project import ProjectStep
 
 
@@ -19,3 +19,24 @@ def extract_step_ids_from_project_steps(
         raise ValueError(f"The following step names do not match any steps: {unmatched_names}")
 
     return [step["id"] for step in matching_steps]
+
+
+def extract_step_id_and_status_filters_from_project_steps(
+    project_steps: list[ProjectStep],
+    step_name_and_status_filters: list[tuple[str, StatusInStep]],
+) -> list[tuple[str, StatusInStep]]:
+    """Convert a list of (step_name, step_status) tuples to (step_id, step_status) tuples."""
+    step_name_to_id = {step["name"]: step["id"] for step in project_steps}
+
+    unmatched_names = [
+        step_name
+        for step_name, _ in step_name_and_status_filters
+        if step_name not in step_name_to_id
+    ]
+    if unmatched_names:
+        raise ValueError(f"The following step names do not match any steps: {unmatched_names}")
+
+    return [
+        (step_name_to_id[step_name], step_status)
+        for step_name, step_status in step_name_and_status_filters
+    ]