feat(lab-4021): add method to activate and deactivate consensus on asset

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

@@ -1,6 +1,6 @@
 """Mixin extending Kili API Gateway class with Asset related operations."""
 
-from typing import Dict, Generator, List
+from typing import Dict, Generator, List, Optional
 
 from kili_formats.tool.annotations_to_json_response import (
     AnnotationsToJsonResponseConverter,
@@ -25,6 +25,7 @@
 )
 from kili.adapters.kili_api_gateway.label.common import get_annotation_fragment
 from kili.adapters.kili_api_gateway.project.common import get_project
+from kili.core.graphql.operations.asset.mutations import GQL_SET_ASSET_CONSENSUS
 from kili.domain.asset import AssetFilters
 from kili.domain.types import ListOrTuple
 
@@ -166,3 +167,26 @@ def count_assets_annotations(self, filters: AssetFilters) -> int:
         count_result = self.graphql_client.execute(GQL_COUNT_ASSET_ANNOTATIONS, payload)
         count: int = count_result["data"]
         return count
+
+    def update_asset_consensus(
+        self,
+        project_id: str,
+        is_consensus: bool,
+        asset_id: Optional[str] = None,
+        external_id: Optional[str] = None,
+    ) -> bool:
+        """Update consensus on an asset."""
+        if asset_id is None and external_id is None:
+            raise ValueError("At least one of asset_id or external_id must be provided")
+
+        payload = {
+            "projectId": project_id,
+            "isConsensus": is_consensus,
+        }
+        if asset_id is not None:
+            payload["assetId"] = asset_id
+        if external_id is not None:
+            payload["externalId"] = external_id
+
+        result = self.graphql_client.execute(GQL_SET_ASSET_CONSENSUS, payload)
+        return result["data"]

src/kili/core/graphql/operations/asset/mutations.py

@@ -27,3 +27,19 @@
   }
 }
 """
+
+GQL_SET_ASSET_CONSENSUS = """
+mutation setAssetConsensus(
+    $assetId: ID,
+    $externalId: String,
+    $projectId: ID!,
+    $isConsensus: Boolean!
+  ) {
+  data: setAssetConsensus(
+    assetId: $assetId,
+    externalId: $externalId,
+    projectId: $projectId,
+    isConsensus: $isConsensus
+  )
+}
+"""

src/kili/domain_api/assets.py

@@ -265,6 +265,7 @@ class AssetsNamespace(DomainNamespace):
     - move_to_next_step(): Move assets to the next workflow step
     - assign(): Assign assets to labelers
     - update_priority(): Update asset priorities
+    - update_consensus(): Activate or deactivate consensus on an asset
 
     Examples:
         >>> kili = Kili()
@@ -2174,3 +2175,78 @@ def update_priority(
             priorities=priorities if priorities is not None else [],
             **kwargs,
         )
+
+    @overload
+    def update_consensus(
+        self,
+        *,
+        asset_id: str,
+        project_id: str,
+        is_consensus: bool,
+    ) -> bool:
+        ...
+
+    @overload
+    def update_consensus(
+        self,
+        *,
+        external_id: str,
+        project_id: str,
+        is_consensus: bool,
+    ) -> bool:
+        ...
+
+    @typechecked
+    def update_consensus(
+        self,
+        *,
+        project_id: str,
+        is_consensus: bool,
+        asset_id: Optional[str] = None,
+        external_id: Optional[str] = None,
+    ) -> bool:
+        """Activate or deactivate consensus on an asset.
+
+        Args:
+            project_id: The project ID.
+            is_consensus: Whether to activate (True) or deactivate (False) consensus on the asset.
+            asset_id: The internal asset ID to modify. Either asset_id or external_id must be provided.
+            external_id: The external ID of the asset to modify. Either asset_id or external_id must be provided.
+
+        Returns:
+            The consensus value that was set (True if consensus was activated, False if deactivated).
+
+        Raises:
+            ValueError: If neither asset_id nor external_id is provided.
+
+        Examples:
+            >>> # Activate consensus on an asset using asset_id
+            >>> result = kili.assets.update_consensus(
+            ...     project_id="my_project",
+            ...     is_consensus=True,
+            ...     asset_id="ckg22d81r0jrg0885unmuswj8"
+            ... )
+            >>> # result is True
+
+            >>> # Activate consensus on an asset using external_id
+            >>> result = kili.assets.update_consensus(
+            ...     project_id="my_project",
+            ...     is_consensus=True,
+            ...     external_id="my_asset_001"
+            ... )
+            >>> # result is True
+
+            >>> # Deactivate consensus on an asset
+            >>> result = kili.assets.update_consensus(
+            ...     project_id="my_project",
+            ...     is_consensus=False,
+            ...     asset_id="ckg22d81r0jrg0885unmuswj8"
+            ... )
+            >>> # result is False
+        """
+        return self._client.update_asset_consensus(
+            project_id=project_id,
+            is_consensus=is_consensus,
+            asset_id=asset_id,
+            external_id=external_id,
+        )

src/kili/presentation/client/asset.py

@@ -764,3 +764,62 @@ def count_assets(
         )
         asset_use_cases = AssetUseCases(self.kili_api_gateway)
         return asset_use_cases.count_assets(filters)
+
+    @typechecked
+    def update_asset_consensus(
+        self,
+        project_id: str,
+        is_consensus: bool,
+        asset_id: Optional[str] = None,
+        external_id: Optional[str] = None,
+    ) -> bool:
+        """Activate or deactivate consensus on an asset.
+
+        Args:
+            project_id: The project ID.
+            is_consensus: Whether to activate (True) or deactivate (False) consensus on the asset.
+            asset_id: The internal asset ID to modify. Either asset_id or external_id must be provided.
+            external_id: The external ID of the asset to modify. Either asset_id or external_id must be provided.
+
+        Returns:
+            The consensus value that was set (True if consensus was activated, False if deactivated).
+
+        Raises:
+            ValueError: If neither asset_id nor external_id is provided.
+
+        Examples:
+            >>> # Activate consensus on an asset using asset_id
+            >>> result = kili.update_asset_consensus(
+            ...     project_id="my_project",
+            ...     is_consensus=True,
+            ...     asset_id="ckg22d81r0jrg0885unmuswj8"
+            ... )
+            >>> # result is True
+
+            >>> # Activate consensus on an asset using external_id
+            >>> result = kili.update_asset_consensus(
+            ...     project_id="my_project",
+            ...     is_consensus=True,
+            ...     external_id="my_asset_001"
+            ... )
+            >>> # result is True
+
+            >>> # Deactivate consensus on an asset
+            >>> result = kili.update_asset_consensus(
+            ...     project_id="my_project",
+            ...     is_consensus=False,
+            ...     asset_id="ckg22d81r0jrg0885unmuswj8"
+            ... )
+            >>> # result is False
+        """
+        if asset_id is None and external_id is None:
+            raise ValueError(
+                "At least one of asset_id or external_id must be provided to update_asset_consensus"
+            )
+
+        return self.kili_api_gateway.update_asset_consensus(
+            project_id=project_id,
+            is_consensus=is_consensus,
+            asset_id=asset_id,
+            external_id=external_id,
+        )