def search_sessions()

in python/src/tablestore_for_agent_memory/memory/memory_store.py [0:0]

• 37 lines of code
• 5 McCabe index (conditional complexity)

    def search_sessions(self,
                        metadata_filter: Optional[Filter] = None,
                        limit: Optional[int] = Field(default=100, le=100, ge=1),
                        next_token: Optional[str] = None
                        ) -> Response:
        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(
            ots_query, limit=limit, get_total_count=False, sort=sort, next_token=next_token
        )
        try:
            search_response = self._client.search(
                table_name=self._session_table_name,
                index_name=self._session_search_index_name,
                search_query=search_query,
                columns_to_get=tablestore.ColumnsToGet(
                    return_type=tablestore.ColumnReturnType.ALL
                ),
            )
        except tablestore.OTSClientError as e:
            logger.exception("tablestore search session failed with client error:%s", e)
            raise e
        except tablestore.OTSServiceError as e:
            logger.exception(
                "tablestore search session 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
        sessions, next_token = TablestoreHelper.search_response_to_sessions(search_response=search_response)
        logger.info(f"tablestore search session index successfully. request_id:[{search_response.request_id}], metadata_filter:[{metadata_filter}], limit:[{limit}], next_token:[{next_token}]")
        return Response(hits=sessions, next_token=next_token)