def search_documents()

in python/src/tablestore_for_agent_memory/knowledge/knowledge_store.py [0:0]

• 49 lines of code
• 7 McCabe index (conditional complexity)

    def search_documents(self,
                         tenant_id: Optional[Union[List[str], str]] = None,
                         metadata_filter: Optional[Filter] = None,
                         limit: Optional[int] = Field(default=100, le=1000, ge=1),
                         next_token: Optional[str] = None,
                         meta_data_to_get: Optional[List[str]] = None,
                         **kwargs: Any,
                         ) -> Response[DocumentHit]:
        if meta_data_to_get is None:
            meta_data_to_get = self._default_meta_data_to_get
        metadata_filter = self._wrap_tenant_id_filter(tenant_id=tenant_id, metadata_filter=metadata_filter)
        ots_query, need_score_sort = TablestoreHelper.paser_search_index_filters(metadata_filter=metadata_filter)
        sort = tablestore.Sort(sorters=[tablestore.ScoreSort(sort_order=tablestore.SortOrder.DESC)]) if need_score_sort else None
        if next_token:
            next_token = base64.b64decode(next_token.encode('utf-8'))
        search_query = tablestore.SearchQuery(
            query=ots_query, limit=limit, get_total_count=False, sort=sort, next_token=next_token
        )
        routing_keys = kwargs.get("routing_keys")
        if routing_keys is None:
            routing_keys = self._build_routing_keys(tenant_id=tenant_id)
        try:
            search_response = self._client.search(
                table_name=self._table_name,
                index_name=self._search_index_name,
                search_query=search_query,
                columns_to_get=tablestore.ColumnsToGet(
                    return_type=tablestore.ColumnReturnType.SPECIFIED,
                    column_names=meta_data_to_get,
                ),
                routing_keys=routing_keys,
            )
        except tablestore.OTSClientError as e:
            logger.exception("tablestore search document failed with client error:%s", e)
            raise e
        except tablestore.OTSServiceError as e:
            logger.exception(
                "tablestore search document failed with error:%s, http_status:%d, error_code:%s, error_message:%s, request_id:%s",
                e,
                e.get_http_status(),
                e.get_error_code(),
                e.get_error_message(),
                e.get_request_id(),
            )
            raise e
        hits, next_token = TablestoreHelper.search_response_to_document(search_response=search_response, text_field=self._text_field, embedding_field=self._embedding_field)
        logger.info(f"tablestore search document index successfully. request_id:[{search_response.request_id}], metadata_filter:[{metadata_filter}], limit:[{limit}], next_token:[{next_token}]")
        response: Response[DocumentHit] = Response(hits=hits, next_token=next_token)
        return response