PYCBC-1550: Add support for scoped search indexes

Motivation
==========
Add support for search index mgmt and search operations from a scope.

Modification
============
* Update C++ core
* Update search index mgmt to use C++ core mgmt API
* Add access to SearchIndexManager from scope instance
* Add access to search() functionality (new path for search API) from scope instance

Change-Id: Ifc572dddff3847da8c67d0bb5c1a73976ec373c0
Reviewed-on: https://review.couchbase.org/c/couchbase-python-client/+/204727
Reviewed-by: Dimitris Christodoulou <dimitris.christodoulou@couchbase.com>
Tested-by: Build Bot <build@couchbase.com>

Author: thejcfactor

acouchbase/management/search.py

@@ -41,7 +41,10 @@
 
 
 class SearchIndexManager(SearchIndexManagerLogic):
-    def __init__(self, connection, loop):
+    def __init__(self,
+                 connection,
+                 loop
+                 ):
         super().__init__(connection)
         self._loop = loop
 
@@ -167,3 +170,141 @@ def get_all_index_stats(self,
                             **kwargs  # type: Dict[str, Any]
                             ) -> Awaitable[Dict[str, Any]]:
         super().get_all_index_stats(*options, **kwargs)
+
+
+class ScopeSearchIndexManager(SearchIndexManagerLogic):
+    """
+    **VOLATILE** This API is subject to change at any time.
+    """
+
+    def __init__(self,
+                 connection,
+                 loop,
+                 bucket_name,  # type: str
+                 scope_name  # type: str
+                 ):
+        super().__init__(connection, bucket_name=bucket_name, scope_name=scope_name)
+        self._loop = loop
+
+    @property
+    def loop(self):
+        """
+        **INTERNAL**
+        """
+        return self._loop
+
+    @AsyncMgmtWrapper.inject_callbacks(None, ManagementType.SearchIndexMgmt, SearchIndexManagerLogic._ERROR_MAPPING)
+    def upsert_index(self,
+                     index,     # type: SearchIndex
+                     *options,  # type: UpsertSearchIndexOptions
+                     **kwargs   # type: Dict[str, Any]
+                     ) -> Awaitable[None]:
+
+        super().upsert_index(index, *options, **kwargs)
+
+    @AsyncMgmtWrapper.inject_callbacks(None, ManagementType.SearchIndexMgmt, SearchIndexManagerLogic._ERROR_MAPPING)
+    def drop_index(self,
+                   index_name,  # type: str
+                   *options,   # type: DropSearchIndexOptions
+                   **kwargs    # type: Dict[str, Any]
+                   ) -> Awaitable[None]:
+
+        super().drop_index(index_name, *options, **kwargs)
+
+    @AsyncMgmtWrapper.inject_callbacks(SearchIndex, ManagementType.SearchIndexMgmt,
+                                       SearchIndexManagerLogic._ERROR_MAPPING)
+    def get_index(self,
+                  index_name,  # type: str
+                  *options,   # type: GetSearchIndexOptions
+                  **kwargs    # type: Dict[str, Any]
+                  ) -> Awaitable[SearchIndex]:
+
+        super().get_index(index_name, *options, **kwargs)
+
+    @AsyncMgmtWrapper.inject_callbacks(SearchIndex, ManagementType.SearchIndexMgmt,
+                                       SearchIndexManagerLogic._ERROR_MAPPING)
+    def get_all_indexes(self,
+                        *options,  # type: GetAllSearchIndexesOptions
+                        **kwargs  # type: Dict[str, Any]
+                        ) -> Awaitable[Iterable[SearchIndex]]:
+        super().get_all_indexes(*options, **kwargs)
+
+    @AsyncMgmtWrapper.inject_callbacks(int, ManagementType.SearchIndexMgmt, SearchIndexManagerLogic._ERROR_MAPPING)
+    def get_indexed_documents_count(self,
+                                    index_name,  # type: str
+                                    *options,   # type: GetSearchIndexedDocumentsCountOptions
+                                    **kwargs    # type: Dict[str, Any]
+                                    ) -> Awaitable[int]:
+        super().get_indexed_documents_count(index_name, *options, **kwargs)
+
+    @AsyncMgmtWrapper.inject_callbacks(None, ManagementType.SearchIndexMgmt, SearchIndexManagerLogic._ERROR_MAPPING)
+    def pause_ingest(self,
+                     index_name,  # type: str
+                     *options,  # type: PauseIngestSearchIndexOptions
+                     **kwargs  # type: Dict[str, Any]
+                     ) -> Awaitable[None]:
+        super().pause_ingest(index_name, *options, **kwargs)
+
+    @AsyncMgmtWrapper.inject_callbacks(None, ManagementType.SearchIndexMgmt, SearchIndexManagerLogic._ERROR_MAPPING)
+    def resume_ingest(self,
+                      index_name,  # type: str
+                      *options,  # type: ResumeIngestSearchIndexOptions
+                      **kwargs  # type: Dict[str, Any]
+                      ) -> Awaitable[None]:
+        super().resume_ingest(index_name, *options, **kwargs)
+
+    @AsyncMgmtWrapper.inject_callbacks(None, ManagementType.SearchIndexMgmt, SearchIndexManagerLogic._ERROR_MAPPING)
+    def allow_querying(self,
+                       index_name,  # type: str
+                       *options,  # type: AllowQueryingSearchIndexOptions
+                       **kwargs  # type: Dict[str, Any]
+                       ) -> Awaitable[None]:
+        super().allow_querying(index_name, *options, **kwargs)
+
+    @AsyncMgmtWrapper.inject_callbacks(None, ManagementType.SearchIndexMgmt, SearchIndexManagerLogic._ERROR_MAPPING)
+    def disallow_querying(self,
+                          index_name,  # type: str
+                          *options,  # type: DisallowQueryingSearchIndexOptions
+                          **kwargs  # type: Dict[str, Any]
+                          ) -> Awaitable[None]:
+        super().disallow_querying(index_name, *options, **kwargs)
+
+    @AsyncMgmtWrapper.inject_callbacks(None, ManagementType.SearchIndexMgmt, SearchIndexManagerLogic._ERROR_MAPPING)
+    def freeze_plan(self,
+                    index_name,  # type: str
+                    *options,  # type: FreezePlanSearchIndexOptions
+                    **kwargs  # type: Dict[str, Any]
+                    ) -> Awaitable[None]:
+        super().freeze_plan(index_name, *options, **kwargs)
+
+    @AsyncMgmtWrapper.inject_callbacks(None, ManagementType.SearchIndexMgmt, SearchIndexManagerLogic._ERROR_MAPPING)
+    def unfreeze_plan(self,
+                      index_name,  # type: str
+                      *options,  # type: UnfreezePlanSearchIndexOptions
+                      **kwargs  # type: Dict[str, Any]
+                      ) -> Awaitable[None]:
+        super().unfreeze_plan(index_name, *options, **kwargs)
+
+    @AsyncMgmtWrapper.inject_callbacks(dict, ManagementType.SearchIndexMgmt, SearchIndexManagerLogic._ERROR_MAPPING)
+    def analyze_document(self,
+                         index_name,  # type: str
+                         document,  # type: Any
+                         *options,  # type: AnalyzeDocumentSearchIndexOptions
+                         **kwargs  # type: Dict[str, Any]
+                         ) -> Awaitable[Dict[str, Any]]:
+        super().analyze_document(index_name, document, *options, **kwargs)
+
+    @AsyncMgmtWrapper.inject_callbacks(dict, ManagementType.SearchIndexMgmt, SearchIndexManagerLogic._ERROR_MAPPING)
+    def get_index_stats(self,
+                        index_name,  # type: str
+                        *options,  # type: GetSearchIndexStatsOptions
+                        **kwargs  # type: Dict[str, Any]
+                        ) -> Awaitable[Dict[str, Any]]:
+        super().get_index_stats(index_name, *options, **kwargs)
+
+    @AsyncMgmtWrapper.inject_callbacks(dict, ManagementType.SearchIndexMgmt, SearchIndexManagerLogic._ERROR_MAPPING)
+    def get_all_index_stats(self,
+                            *options,  # type: GetAllSearchIndexStatsOptions
+                            **kwargs  # type: Dict[str, Any]
+                            ) -> Awaitable[Dict[str, Any]]:
+        super().get_all_index_stats(*options, **kwargs)

acouchbase/scope.py

@@ -16,10 +16,12 @@
 from typing import (TYPE_CHECKING,
                     Any,
                     Awaitable,
+                    Dict,
                     Optional)
 
 from acouchbase.analytics import AnalyticsQuery, AsyncAnalyticsRequest
 from acouchbase.collection import Collection
+from acouchbase.management.search import ScopeSearchIndexManager
 from acouchbase.n1ql import AsyncN1QLRequest, N1QLQuery
 from acouchbase.search import AsyncFullTextSearchRequest, SearchQueryBuilder
 from couchbase.options import (AnalyticsOptions,
@@ -28,10 +30,11 @@
 from couchbase.result import (AnalyticsResult,
                               QueryResult,
                               SearchResult)
+from couchbase.serializer import Serializer
 from couchbase.transcoder import Transcoder
 
 if TYPE_CHECKING:
-    from acouchbase.search import SearchQuery
+    from couchbase.search import SearchQuery, SearchRequest
 
 
 class AsyncScope:
@@ -65,6 +68,17 @@ def loop(self):
         """
         return self._bucket.loop
 
+    @property
+    def default_serializer(self) -> Optional[Serializer]:
+        return self._bucket.default_serializer
+
+    @property
+    def streaming_timeouts(self):
+        """
+        **INTERNAL**
+        """
+        return self._bucket.streaming_timeouts
+
     @property
     def default_transcoder(self) -> Optional[Transcoder]:
         return self._bucket.default_transcoder
@@ -107,12 +121,11 @@ def _set_connection(self):
         """
         self._connection = self._bucket.connection
 
-    def query(
-        self,
-        statement,  # type: str
-        *options,  # type: QueryOptions
-        **kwargs  # type: Any
-    ) -> QueryResult:
+    def query(self,
+              statement,  # type: str
+              *options,  # type: QueryOptions
+              **kwargs   # type: Dict[str, Any]
+              ) -> QueryResult:
         """Executes a N1QL query against the scope.
 
         .. note::
@@ -203,12 +216,11 @@ def query(
                                                                   query.params,
                                                                   **request_args))
 
-    def analytics_query(
-        self,
-        statement,  # type: str
-        *options,  # type: AnalyticsOptions
-        **kwargs
-    ) -> AnalyticsResult:
+    def analytics_query(self,
+                        statement,  # type: str
+                        *options,  # type: AnalyticsOptions
+                        **kwargs   # type: Dict[str, Any]
+                        ) -> AnalyticsResult:
         """Executes an analaytics query against the scope.
 
         .. note::
@@ -299,13 +311,12 @@ def analytics_query(
                                                                                 query.params,
                                                                                 **request_args))
 
-    def search_query(
-        self,
-        index,  # type: str
-        query,  # type: SearchQuery
-        *options,  # type: SearchOptions
-        **kwargs
-    ) -> SearchResult:
+    def search_query(self,
+                     index,  # type: str
+                     query,  # type: SearchQuery
+                     *options,  # type: SearchOptions
+                     **kwargs   # type: Dict[str, Any]
+                     ) -> SearchResult:
         """Executes an search query against the scope.
 
         .. note::
@@ -409,14 +420,47 @@ def search_query(
         if not ('scope_name' in opt or 'scope_name' in kwargs):
             kwargs['scope_name'] = f'{self.name}'
 
-        query = SearchQueryBuilder.create_search_query_object(
-            index, query, *options, **kwargs
-        )
+        query = SearchQueryBuilder.create_search_query_object(index, query, *options, **kwargs)
         return SearchResult(AsyncFullTextSearchRequest.generate_search_request(self.connection,
                                                                                self.loop,
                                                                                query.as_encodable(),
                                                                                **request_args))
 
+    def search(self,
+               index,  # type: str
+               request,  # type: SearchRequest
+               *options,  # type: SearchOptions
+               **kwargs,  # type: Dict[str, Any]
+               ) -> SearchResult:
+        """
+        **VOLATILE** This API is subject to change at any time.
+        """
+        # See couchbase.cluster.search() for note on streaming timeout
+        streaming_timeout = self.streaming_timeouts.get('search_timeout', None)
+        query = SearchQueryBuilder.create_search_query_from_request(index, request, *options, **kwargs)
+        req = AsyncFullTextSearchRequest.generate_search_request(self.connection,
+                                                                 self.loop,
+                                                                 query.as_encodable(),
+                                                                 default_serializer=self.default_serializer,
+                                                                 streaming_timeout=streaming_timeout,
+                                                                 bucket_name=self.bucket_name,
+                                                                 scope_name=self.name)
+        return SearchResult(req)
+
+    def search_indexes(self) -> ScopeSearchIndexManager:
+        """
+        **VOLATILE** This API is subject to change at any time.
+
+        Get a :class:`~acouchbase.management.search.ScopeSearchIndexManager` which can be used to manage the search
+        indexes of this scope.
+
+        Returns:
+            :class:`~acouchbase.management.search.ScopeSearchIndexManager`: A :class:`~acouchbase.management.search.ScopeSearchIndexManager` instance.
+
+        """  # noqa: E501
+        # TODO:  AlreadyShutdownException?
+        return ScopeSearchIndexManager(self.connection, self.loop, self.bucket_name, self.name)
+
     @staticmethod
     def default_name():
         return "_default"

acouchbase/tests/_test_utils.py

@@ -13,6 +13,8 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
+from __future__ import annotations
+
 import asyncio
 from typing import (Any,
                     Callable,
@@ -112,6 +114,7 @@ def __init__(self, cluster, bucket, collection, cluster_config, **kwargs):
         self._collection_spec = None
         self._scope = None
         self._named_collection = None
+        self._use_scope_search_mgmt = False
 
     @property
     def collection(self) -> Collection:
@@ -182,6 +185,10 @@ def rate_limit_params(self) -> Optional[RateLimitData]:
         """Returns the rate limit testing data"""
         return self._rate_limit_params if hasattr(self, '_rate_limit_params') else None
 
+    @property
+    def use_scope_search_mgmt(self) -> bool:
+        return self._use_scope_search_mgmt
+
     @classmethod  # noqa: C901
     async def get_environment(cls,  # noqa: C901
                               test_suite,
@@ -239,6 +246,33 @@ async def get_environment(cls,  # noqa: C901
 
         return cb_env
 
+    def disable_scope_search_mgmt(self) -> TestEnvironment:
+        self._use_scope_search_mgmt = False
+        return self
+
+    def disable_search_mgmt(self) -> TestEnvironment:
+        self.check_if_feature_supported('search_index_mgmt')
+        if not hasattr(self, '_sixm'):
+            del self._sixm
+
+        return self
+
+    def enable_scope_search_mgmt(self) -> TestEnvironment:
+        self._use_scope_search_mgmt = True
+        return self
+
+    def enable_search_mgmt(self) -> TestEnvironment:
+        self.check_if_feature_supported('search_index_mgmt')
+        if self.use_scope_search_mgmt:
+            if not hasattr(self.scope, 'search_indexes'):
+                pytest.skip('Search index management not available on scope.')
+            self._sixm = self.scope.search_indexes()
+        else:
+            if not hasattr(self.cluster, 'search_indexes'):
+                pytest.skip('Search index management not available on cluster.')
+            self._sixm = self.cluster.search_indexes()
+        return self
+
     async def get_new_key_value(self, reset=True):
         if reset is True:
             try: