Merge pull request #116 from pinecone-io/list-canopy-indexes

support list canopy indexes

src/canopy/knowledge_base/__init__.py

@@ -1 +1,3 @@
+from .knowledge_base import connect_to_pinecone
+from .knowledge_base import list_canopy_indexes
 from .knowledge_base import KnowledgeBase

src/canopy/knowledge_base/knowledge_base.py

@@ -37,6 +37,37 @@
 DELETE_STARTER_CHUNKS_PER_DOC = 32
 
 
+def connect_to_pinecone():
+    """
+    Connect to Pinecone.
+    This method is called automatically when creating a new KnowledgeBase object.
+    Or when calling `list_canopy_indexes()`.
+    """
+    try:
+        pinecone_init()
+        pinecone_whoami()
+    except Exception as e:
+        raise RuntimeError("Failed to connect to Pinecone. "
+                           "Please check your credentials and try again") from e
+
+
+def list_canopy_indexes() -> List[str]:
+    """
+    List all Canopy indexes in the current Pinecone account.
+
+    Example:
+        >>> from canopy.knowledge_base import list_canopy_indexes
+        >>> list_canopy_indexes()
+            ['canopy--my_index', 'canopy--my_index2']
+
+    Returns:
+        A list of Canopy index names.
+    """
+
+    connect_to_pinecone()
+    return [index for index in list_indexes() if index.startswith(INDEX_NAME_PREFIX)]
+
+
 class KnowledgeBase(BaseKnowledgeBase):
 
     """
@@ -160,20 +191,11 @@ def __init__(self,
         # `create_canopy_index()`
         self._index: Optional[Index] = None
 
-    @staticmethod
-    def _connect_pinecone():
-        try:
-            pinecone_init()
-            pinecone_whoami()
-        except Exception as e:
-            raise RuntimeError("Failed to connect to Pinecone. "
-                               "Please check your credentials and try again") from e
-
     def _connect_index(self,
                        connect_pinecone: bool = True
                        ) -> None:
         if connect_pinecone:
-            self._connect_pinecone()
+            connect_to_pinecone()
 
         if self.index_name not in list_indexes():
             raise RuntimeError(
@@ -297,7 +319,7 @@ def create_canopy_index(self,
                                  "Please provide the vectors' dimension")
 
         # connect to pinecone and create index
-        self._connect_pinecone()
+        connect_to_pinecone()
 
         if self.index_name in list_indexes():
             raise RuntimeError(

src/canopy_cli/cli.py

@@ -18,6 +18,7 @@
 from urllib.parse import urljoin
 
 from canopy.knowledge_base import KnowledgeBase
+from canopy.knowledge_base import connect_to_pinecone
 from canopy.models.data_models import Document
 from canopy.tokenizer import Tokenizer
 from canopy_cli.data_loader import (
@@ -71,7 +72,7 @@ def wait_for_service(chat_service_url: str):
 
 def validate_connection():
     try:
-        KnowledgeBase._connect_pinecone()
+        connect_to_pinecone()
     except RuntimeError as e:
         msg = (
             f"{str(e)}\n"

tests/system/knowledge_base/test_knowledge_base.py

@@ -12,7 +12,7 @@
 )
 from dotenv import load_dotenv
 from datetime import datetime
-from canopy.knowledge_base import KnowledgeBase
+from canopy.knowledge_base import KnowledgeBase, list_canopy_indexes
 from canopy.knowledge_base.chunker import Chunker
 from canopy.knowledge_base.knowledge_base import INDEX_NAME_PREFIX
 from canopy.knowledge_base.models import DocumentWithScore
@@ -226,6 +226,15 @@ def test_create_index(index_full_name, knowledge_base):
     assert knowledge_base._index.describe_index_stats()
 
 
+def test_list_indexes(index_full_name):
+    actual = list_canopy_indexes()
+    expected = [index for index in pinecone.list_indexes()
+                if index.startswith(INDEX_NAME_PREFIX)]
+
+    assert index_full_name in actual
+    assert sorted(actual) == sorted(expected)
+
+
 def test_is_verify_index_connection_happy_path(knowledge_base):
     knowledge_base.verify_index_connection()