feat: add support for system indexes [ECS-1791] (#1605)

Author: gcuip

packages/uipath-platform/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath-platform"
-version = "0.1.41"
+version = "0.1.42"
 description = "HTTP client library for programmatic access to UiPath Platform"
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"

packages/uipath-platform/src/uipath/platform/context_grounding/_context_grounding_service.py

@@ -19,6 +19,7 @@
 from ..errors import (
     BatchTransformFailedException,
     BatchTransformNotCompleteException,
+    ContextGroundingIndexNotFoundError,
     IngestionInProgressException,
     UnsupportedDataSourceException,
 )
@@ -256,37 +257,81 @@ async def retrieve_across_folders_async(
             ContextGroundingIndex.model_validate(item) for item in response["value"]
         ]
 
+    @traced(name="contextgrounding_retrieve_system_indexes", run_type="uipath")
+    def _retrieve_system_indexes(
+        self,
+        name: Optional[str] = None,
+    ) -> List[ContextGroundingIndex]:
+        spec = self._retrieve_system_indexes_spec(name=name)
+
+        response = self.request(
+            spec.method,
+            spec.endpoint,
+            params=spec.params,
+        ).json()
+
+        return [
+            ContextGroundingIndex.model_validate(item) for item in response["value"]
+        ]
+
+    @traced(name="contextgrounding_retrieve_system_indexes", run_type="uipath")
+    async def _retrieve_system_indexes_async(
+        self,
+        name: Optional[str] = None,
+    ) -> List[ContextGroundingIndex]:
+        spec = self._retrieve_system_indexes_spec(name=name)
+
+        response = (
+            await self.request_async(
+                spec.method,
+                spec.endpoint,
+                params=spec.params,
+            )
+        ).json()
+
+        return [
+            ContextGroundingIndex.model_validate(item) for item in response["value"]
+        ]
+
     @resource_override(resource_type="index")
     @traced(name="contextgrounding_retrieve", run_type="uipath")
     def retrieve(
         self,
         name: str,
         folder_key: Optional[str] = None,
         folder_path: Optional[str] = None,
+        include_system_indexes: bool = False,
     ) -> ContextGroundingIndex:
         """Retrieve context grounding index information by its name.
 
         If no folder_key or folder_path is provided and no folder context is
-        configured, falls back to searching across all folders.
+        configured, falls back to searching across all folders. When
+        ``include_system_indexes`` is True, an additional fallback against
+        system indexes is attempted before raising not-found.
 
         Args:
             name (str): The name of the context index to retrieve.
             folder_key (Optional[str]): The key of the folder where the index resides.
             folder_path (Optional[str]): The path of the folder where the index resides.
+            include_system_indexes (bool): If True, fall back to system indexes
+                when the index is not found in the per-folder or across-folders listings.
+                Defaults to False.
 
         Returns:
             ContextGroundingIndex: The index information, including its configuration and metadata if found.
 
         Raises:
-            Exception: If no index with the given name is found.
+            ContextGroundingIndexNotFoundError: If no index with the given name is found.
         """
         resolved_folder_key = self._resolve_folder_key(folder_key, folder_path)
         if not resolved_folder_key:
             indexes = self.retrieve_across_folders(name=name)
             try:
                 return next(index for index in indexes if index.name == name)
-            except StopIteration as e:
-                raise Exception("ContextGroundingIndex not found") from e
+            except StopIteration:
+                if include_system_indexes:
+                    return self._retrieve_from_system_indexes(name)
+                raise ContextGroundingIndexNotFoundError(name) from None
 
         spec = self._retrieve_spec(
             name,
@@ -305,8 +350,10 @@ def retrieve(
                 for item in response["value"]
                 if item["name"] == name
             )
-        except StopIteration as e:
-            raise Exception("ContextGroundingIndex not found") from e
+        except StopIteration:
+            if include_system_indexes:
+                return self._retrieve_from_system_indexes(name)
+            raise ContextGroundingIndexNotFoundError(name) from None
 
     @resource_override(resource_type="index")
     @traced(name="contextgrounding_retrieve", run_type="uipath")
@@ -315,30 +362,38 @@ async def retrieve_async(
         name: str,
         folder_key: Optional[str] = None,
         folder_path: Optional[str] = None,
+        include_system_indexes: bool = False,
     ) -> ContextGroundingIndex:
         """Asynchronously retrieve context grounding index information by its name.
 
         If no folder_key or folder_path is provided and no folder context is
-        configured, falls back to searching across all folders.
+        configured, falls back to searching across all folders. When
+        ``include_system_indexes`` is True, an additional fallback against
+        system indexes is attempted before raising not-found.
 
         Args:
             name (str): The name of the context index to retrieve.
             folder_key (Optional[str]): The key of the folder where the index resides.
             folder_path (Optional[str]): The path of the folder where the index resides.
+            include_system_indexes (bool): If True, fall back to system indexes when
+                the index is not found in the per-folder or across-folders listings.
+                Defaults to False.
 
         Returns:
             ContextGroundingIndex: The index information, including its configuration and metadata if found.
 
         Raises:
-            Exception: If no index with the given name is found.
+            ContextGroundingIndexNotFoundError: If no index with the given name is found.
         """
         resolved_folder_key = self._resolve_folder_key(folder_key, folder_path)
         if not resolved_folder_key:
             indexes = await self.retrieve_across_folders_async(name=name)
             try:
                 return next(index for index in indexes if index.name == name)
-            except StopIteration as e:
-                raise Exception("ContextGroundingIndex not found") from e
+            except StopIteration:
+                if include_system_indexes:
+                    return await self._retrieve_from_system_indexes_async(name)
+                raise ContextGroundingIndexNotFoundError(name) from None
 
         spec = self._retrieve_spec(
             name,
@@ -359,8 +414,26 @@ async def retrieve_async(
                 for item in response["value"]
                 if item["name"] == name
             )
-        except StopIteration as e:
-            raise Exception("ContextGroundingIndex not found") from e
+        except StopIteration:
+            if include_system_indexes:
+                return await self._retrieve_from_system_indexes_async(name)
+            raise ContextGroundingIndexNotFoundError(name) from None
+
+    def _retrieve_from_system_indexes(self, name: str) -> ContextGroundingIndex:
+        indexes = self._retrieve_system_indexes(name=name)
+        try:
+            return next(index for index in indexes if index.name == name)
+        except StopIteration:
+            raise ContextGroundingIndexNotFoundError(name) from None
+
+    async def _retrieve_from_system_indexes_async(
+        self, name: str
+    ) -> ContextGroundingIndex:
+        indexes = await self._retrieve_system_indexes_async(name=name)
+        try:
+            return next(index for index in indexes if index.name == name)
+        except StopIteration:
+            raise ContextGroundingIndexNotFoundError(name) from None
 
     @traced(name="contextgrounding_list", run_type="uipath")
     def list(
@@ -1489,6 +1562,7 @@ def unified_search(
         scope: Optional[UnifiedSearchScope] = None,
         folder_key: Optional[str] = None,
         folder_path: Optional[str] = None,
+        include_system_indexes: bool = False,
     ) -> UnifiedQueryResult:
         """Perform a unified search on a context grounding index.
 
@@ -1504,11 +1578,19 @@ def unified_search(
             scope (Optional[UnifiedSearchScope]): Optional search scope (folder, extension).
             folder_key (Optional[str]): The key of the folder where the index resides.
             folder_path (Optional[str]): The path of the folder where the index resides.
+            include_system_indexes (bool): If True, fall back to tenant-wide
+                system indexes when the index is not found in folder or
+                across-folders listings. Defaults to False.
 
         Returns:
             UnifiedQueryResult: The unified search result containing semantic and/or tabular results.
         """
-        index = self.retrieve(name, folder_key=folder_key, folder_path=folder_path)
+        index = self.retrieve(
+            name,
+            folder_key=folder_key,
+            folder_path=folder_path,
+            include_system_indexes=include_system_indexes,
+        )
 
         folder_key = folder_key or index.folder_key
 
@@ -1544,6 +1626,7 @@ async def unified_search_async(
         scope: Optional[UnifiedSearchScope] = None,
         folder_key: Optional[str] = None,
         folder_path: Optional[str] = None,
+        include_system_indexes: bool = False,
     ) -> UnifiedQueryResult:
         """Asynchronously perform a unified search on a context grounding index.
 
@@ -1559,12 +1642,18 @@ async def unified_search_async(
             scope (Optional[UnifiedSearchScope]): Optional search scope (folder, extension).
             folder_key (Optional[str]): The key of the folder where the index resides.
             folder_path (Optional[str]): The path of the folder where the index resides.
+            include_system_indexes (bool): If True, fall back to tenant-wide
+                system indexes when the index is not found in folder or
+                across-folders listings. Defaults to False.
 
         Returns:
             UnifiedQueryResult: The unified search result containing semantic and/or tabular results.
         """
         index = await self.retrieve_async(
-            name, folder_key=folder_key, folder_path=folder_path
+            name,
+            folder_key=folder_key,
+            folder_path=folder_path,
+            include_system_indexes=include_system_indexes,
         )
         if index and index.in_progress_ingestion():
             raise IngestionInProgressException(index_name=name)
@@ -1911,6 +2000,16 @@ def _ingest_spec(
             },
         )
 
+    @staticmethod
+    def _odata_name_filter(name: str) -> str:
+        """Build an OData ``Name eq '<name>'`` filter with single quotes escaped.
+
+        OData string literals escape ``'`` by doubling it. URL encoding of the
+        resulting filter is handled by the HTTP client when params are passed
+        as a dict.
+        """
+        return "Name eq '{}'".format(name.replace("'", "''"))
+
     def _retrieve_across_folders_spec(
         self,
         name: Optional[str] = None,
@@ -1919,14 +2018,30 @@ def _retrieve_across_folders_spec(
             "$expand": "dataSource",
         }
         if name:
-            params["$filter"] = f"Name eq '{name}'"
+            params["$filter"] = self._odata_name_filter(name)
 
         return RequestSpec(
             method="GET",
             endpoint=Endpoint("/ecs_/v2/indexes/allacrossfolders"),
             params=params,
         )
 
+    def _retrieve_system_indexes_spec(
+        self,
+        name: Optional[str] = None,
+    ) -> RequestSpec:
+        params: Dict[str, str] = {
+            "$expand": "dataSource",
+        }
+        if name:
+            params["$filter"] = self._odata_name_filter(name)
+
+        return RequestSpec(
+            method="GET",
+            endpoint=Endpoint("/ecs_/v2/indexes/allsystemindexes"),
+            params=params,
+        )
+
     def _list_spec(
         self,
         folder_key: Optional[str] = None,
@@ -1954,7 +2069,7 @@ def _retrieve_spec(
             method="GET",
             endpoint=Endpoint("/ecs_/v2/indexes"),
             params={
-                "$filter": f"Name eq '{name}'",
+                "$filter": self._odata_name_filter(name),
                 "$expand": "dataSource",
             },
             headers={

packages/uipath-platform/src/uipath/platform/errors/__init__.py

@@ -8,6 +8,7 @@
 - FolderNotFoundException: Raised when a folder cannot be found
 - UnsupportedDataSourceException: Raised when an operation is attempted on an unsupported data source type
 - IngestionInProgressException: Raised when a search is attempted on an index during ingestion
+- ContextGroundingIndexNotFoundError: Raised when a context grounding index cannot be resolved by name
 - BatchTransformFailedException: Raised when a batch transform has failed
 - BatchTransformNotCompleteException: Raised when attempting to get results from an incomplete batch transform
 - OperationNotCompleteException: Raised when attempting to get results from an incomplete operation
@@ -18,6 +19,9 @@
 from ._base_url_missing_error import BaseUrlMissingError
 from ._batch_transform_failed_exception import BatchTransformFailedException
 from ._batch_transform_not_complete_exception import BatchTransformNotCompleteException
+from ._context_grounding_index_not_found_exception import (
+    ContextGroundingIndexNotFoundError,
+)
 from ._enriched_exception import EnrichedException, ExtractedErrorInfo
 from ._folder_not_found_exception import FolderNotFoundException
 from ._ingestion_in_progress_exception import IngestionInProgressException
@@ -30,6 +34,7 @@
     "BaseUrlMissingError",
     "BatchTransformFailedException",
     "BatchTransformNotCompleteException",
+    "ContextGroundingIndexNotFoundError",
     "EnrichedException",
     "ExtractedErrorInfo",
     "FolderNotFoundException",

packages/uipath-platform/src/uipath/platform/errors/_context_grounding_index_not_found_exception.py

@@ -0,0 +1,13 @@
+from typing import Optional
+
+
+class ContextGroundingIndexNotFoundError(Exception):
+    """Raised when a context grounding index cannot be resolved by name."""
+
+    def __init__(self, index_name: Optional[str] = None):
+        self.index_name = index_name
+        if index_name:
+            self.message = f"ContextGroundingIndex '{index_name}' not found"
+        else:
+            self.message = "ContextGroundingIndex not found"
+        super().__init__(self.message)